Add integrity checking for downloaded files

OpCore-Simplify.py

@@ -276,6 +276,9 @@ class OCPE:
             if not tool_path in tool_loaded:
                 files_to_remove.append(os.path.join(tools_directory, tool_path))
 
+        if "manifest.json" in os.listdir(self.result_dir):
+            files_to_remove.append(os.path.join(self.result_dir, "manifest.json"))
+
         for file_path in files_to_remove:
             try:
                 if os.path.isdir(file_path):
@@ -421,7 +424,12 @@ class OCPE:
                     self.s.smbios_specific_options(customized_hardware, smbios_model, macos_version, self.ac.patches, self.k)
                     continue
 
-                if not self.o.gather_bootloader_kexts(self.k.kexts, macos_version):
+                try:
+                    self.o.gather_bootloader_kexts(self.k.kexts, macos_version)
+                except Exception as e:
+                    print("\033[91mError: {}\033[0m".format(e))
+                    print("")
+                    self.u.request_input("Press Enter to continue...")
                     continue
                 
                 self.build_opencore_efi(customized_hardware, disabled_devices, smbios_model, macos_version, needs_oclp)

Scripts/gathering_files.py

@@ -1,5 +1,6 @@
 from Scripts import github
 from Scripts import kext_maestro
+from Scripts import integrity_checker
 from Scripts import resource_fetcher
 from Scripts import utils
 import os
@@ -15,6 +16,7 @@ class gatheringFiles:
         self.github = github.Github()
         self.kext = kext_maestro.KextMaestro()
         self.fetcher = resource_fetcher.ResourceFetcher()
+        self.integrity_checker = integrity_checker.IntegrityChecker()
         self.dortania_builds_url = "https://raw.githubusercontent.com/dortania/build-repo/builds/latest.json"
         self.ocbinarydata_url = "https://github.com/acidanthera/OcBinaryData/archive/refs/heads/master.zip"
         self.amd_vanilla_patches_url = "https://raw.githubusercontent.com/AMD-OSX/AMD_Vanilla/beta/patches.plist"
@@ -22,7 +24,6 @@ class gatheringFiles:
         self.hyper_threading_patches_url = "https://github.com/b00t0x/CpuTopologyRebuild/raw/refs/heads/master/patches_ht.plist"
         self.temporary_dir = self.utils.get_temporary_dir()
         self.ock_files_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "OCK_Files")
-        self.bootloader_kexts_data_path = os.path.join(os.path.dirname(os.path.dirname(os.path.realpath(__file__))), "bootloader_kexts_data.json")
         self.download_history_file = os.path.join(self.ock_files_dir, "history.json")
 
     def get_product_index(self, product_list, product_name_name):
@@ -31,58 +32,55 @@ class gatheringFiles:
                 return index
         return None
         
-    def get_bootloader_kexts_data(self, kexts):
-        download_urls = self.utils.read_file(self.bootloader_kexts_data_path)
-
-        if not isinstance(download_urls, list):
-            download_urls = []
-
+    def update_download_database(self, kexts, download_history):
         dortania_builds_data = self.fetcher.fetch_and_parse_content(self.dortania_builds_url, "json")
         seen_repos = set()
 
-        def add_product_to_download_urls(products):
+        def add_product_to_download_database(products):
             if isinstance(products, dict):
                 products = [products]
 
             for product in products:
-                product_index = self.get_product_index(download_urls, product.get("product_name"))
+                if not product or not product.get("product_name"):
+                    continue
+
+                product_index = self.get_product_index(download_history, product.get("product_name"))
 
                 if product_index is None:
-                    download_urls.append(product)
+                    download_history.append(product)
                 else:
-                    download_urls[product_index] = product
+                    download_history[product_index].update(product)
 
         for kext in kexts:
             if not kext.checked:
                 continue
 
             if kext.download_info:
-                add_product_to_download_urls({"product_name": kext.name, **kext.download_info})
+                if not kext.download_info.get("sha256"):
+                    kext.download_info["sha256"] = None
+                add_product_to_download_database({"product_name": kext.name, **kext.download_info})
             elif kext.github_repo and kext.github_repo.get("repo") not in seen_repos:
                 name = kext.github_repo.get("repo")
                 seen_repos.add(name)
                 if name in dortania_builds_data:
-                    add_product_to_download_urls({
+                    add_product_to_download_database({
                         "product_name": name,
                         "id": dortania_builds_data[name]["versions"][0]["release"]["id"], 
-                        "url": dortania_builds_data[name]["versions"][0]["links"]["release"]
+                        "url": dortania_builds_data[name]["versions"][0]["links"]["release"],
+                        "sha256": dortania_builds_data[name]["versions"][0]["hashes"]["release"]["sha256"]
                     })
                 else:
                     latest_release = self.github.get_latest_release(kext.github_repo.get("owner"), kext.github_repo.get("repo")) or {}
-                    add_product_to_download_urls(latest_release.get("assets"))
+                    add_product_to_download_database(latest_release.get("assets"))
 
-        add_product_to_download_urls({
+        add_product_to_download_database({
             "product_name": "OpenCorePkg",
             "id": dortania_builds_data["OpenCorePkg"]["versions"][0]["release"]["id"], 
-            "url": dortania_builds_data["OpenCorePkg"]["versions"][0]["links"]["release"]
+            "url": dortania_builds_data["OpenCorePkg"]["versions"][0]["links"]["release"],
+            "sha256": dortania_builds_data["OpenCorePkg"]["versions"][0]["hashes"]["release"]["sha256"]
         })
 
-        sorted_download_urls = sorted(download_urls, key=lambda x:x["product_name"])
-
-        self.utils.create_folder(self.ock_files_dir)
-        self.utils.write_file(self.bootloader_kexts_data_path, sorted_download_urls)
-    
-        return sorted_download_urls
+        return sorted(download_history, key=lambda x:x["product_name"])
     
     def move_bootloader_kexts_to_product_directory(self, product_name):
         if not os.path.exists(self.temporary_dir):
@@ -145,11 +143,10 @@ class gatheringFiles:
         print("Please wait for download OpenCorePkg, kexts and macserial...")
 
         download_history = self.utils.read_file(self.download_history_file)
-
         if not isinstance(download_history, list):
             download_history = []
 
-        bootloader_kext_urls = self.get_bootloader_kexts_data(kexts)
+        download_database = self.update_download_database(kexts, download_history)
         
         self.utils.create_folder(self.temporary_dir)
 
@@ -183,104 +180,95 @@ class gatheringFiles:
             elif product_name == "UTBDefault":
                 product_name = "USBToolBox"
 
-            product_download_index = self.get_product_index(bootloader_kext_urls, product_name)
+            product_download_index = self.get_product_index(download_database, product_name)
             if product_download_index is None:
-                if product.github_repo:
-                    product_download_index = self.get_product_index(bootloader_kext_urls, product.github_repo.get("repo"))
+                if hasattr(product, 'github_repo') and product.github_repo:
+                    product_download_index = self.get_product_index(download_database, product.github_repo.get("repo"))
             
-            if product_download_index is not None:
-                _, product_id, product_download_url = bootloader_kext_urls[product_download_index].values()
+            if product_download_index is None:
+                print("\n")
+                print("Could not find download URL for {}.".format(product_name))
+                continue
+
+            product_info = download_database[product_download_index]
+            product_id = product_info.get("id")
+            product_download_url = product_info.get("url")
+            sha256_hash = product_info.get("sha256")
 
             if product_download_url in seen_download_urls:
                 continue
             seen_download_urls.add(product_download_url)
-            else:
-                product_id = product_download_url = None
 
             product_history_index = self.get_product_index(download_history, product_name)
+            asset_dir = os.path.join(self.ock_files_dir, product_name)
+            manifest_path = os.path.join(asset_dir, "manifest.json")
+
+            if product_history_index is not None:
+                history_item = download_history[product_history_index]
+                is_latest_id = (product_id == history_item.get("id"))
+                folder_is_valid, _ = self.integrity_checker.verify_folder_integrity(asset_dir, manifest_path)
+                
+                if is_latest_id and folder_is_valid:
+                    print(f"\nLatest version of {product_name} already downloaded.")
+                    continue
 
             print("")
-            if product_history_index is None:
-                print("Please wait for download {}...".format(product_name))
-            else:
-                if product_id == download_history[product_history_index].get("id"):
-                    print("Latest version of {} already downloaded.".format(product_name))
-                    continue
-                else:
-                    print("Updating {}...".format(product_name))
-
-            if product_download_url is not None:
-                print("from " + product_download_url)
+            print("Updating" if product_history_index is not None else "Please wait for download", end=" ")
+            print("{}...".format(product_name))
+            print("")
+            if product_download_url:
+                print("from {}".format(product_download_url))
+                print("")
             else:
+                print("")
                 print("Could not find download URL for {}.".format(product_name))
-                print("Please try again later.")
                 print("")
                 self.utils.request_input()
                 shutil.rmtree(self.temporary_dir, ignore_errors=True)
                 return False
-            print("")
 
             zip_path = os.path.join(self.temporary_dir, product_name) + ".zip"
-            self.fetcher.download_and_save_file(product_download_url, zip_path)
-
-            if not os.path.exists(zip_path):
-                if product_history_index is not None:
-                    print("Using previously version of {}.".format(product_name))
+            if not self.fetcher.download_and_save_file(product_download_url, zip_path, sha256_hash):
+                folder_is_valid, _ = self.integrity_checker.verify_folder_integrity(asset_dir, manifest_path)
+                if product_history_index is not None and folder_is_valid:
+                    print("Using previously downloaded version of {}.".format(product_name))
                     continue
                 else:
-                    print("")
-                    print("Could not download {} at this time.".format(product_name))
-                    print("Please try again later.")
-                    print("")
-                    self.utils.request_input()
-                    shutil.rmtree(self.temporary_dir, ignore_errors=True)
-                    return False
+                    raise Exception("Could not download {} at this time. Please try again later.".format(product_name))
             
             self.utils.extract_zip_file(zip_path)
-
-            asset_dir = os.path.join(self.ock_files_dir, product_name)
             self.utils.create_folder(asset_dir, remove_content=True)
             
             while True:
-                zip_files = self.utils.find_matching_paths(os.path.join(self.temporary_dir, product_name), extension_filter=".zip")
-
-                if not zip_files:
+                nested_zip_files = self.utils.find_matching_paths(os.path.join(self.temporary_dir, product_name), extension_filter=".zip")
+                if not nested_zip_files:
                     break
-
-                for zip_file, file_type in zip_files:
+                for zip_file, _ in nested_zip_files:
                     full_zip_path = os.path.join(self.temporary_dir, product_name, zip_file)
                     self.utils.extract_zip_file(full_zip_path)
                     os.remove(full_zip_path)
 
             if "OpenCore" in product_name:
-                zip_path = os.path.join(self.temporary_dir, "OcBinaryData.zip")
+                oc_binary_data_zip_path = os.path.join(self.temporary_dir, "OcBinaryData.zip")
                 print("")
                 print("Please wait for download OcBinaryData...")
-                print("from " + self.ocbinarydata_url)
+                print("from {}".format(self.ocbinarydata_url))
                 print("")
-                self.fetcher.download_and_save_file(self.ocbinarydata_url, zip_path)
+                self.fetcher.download_and_save_file(self.ocbinarydata_url, oc_binary_data_zip_path)
 
-                if not os.path.exists(zip_path):
+                if not os.path.exists(oc_binary_data_zip_path):
                     print("")
                     print("Could not download OcBinaryData at this time.")
-                    print("Please try again later.")
-                    print("")
+                    print("Please try again later.\n")
                     self.utils.request_input()
                     shutil.rmtree(self.temporary_dir, ignore_errors=True)
                     return False
                 
-                self.utils.extract_zip_file(zip_path)
+                self.utils.extract_zip_file(oc_binary_data_zip_path)
 
             if self.move_bootloader_kexts_to_product_directory(product_name):
-                if product_history_index is None:
-                    download_history.append({
-                        "product_name": product_name, 
-                        "id": product_id
-                    })
-                else:
-                    download_history[product_history_index]["id"] = product_id
-                
-                self.utils.write_file(self.download_history_file, download_history)
+                self.integrity_checker.generate_folder_manifest(asset_dir, manifest_path)
+                self._update_download_history(download_history, product_name, product_id, product_download_url, sha256_hash)
 
         shutil.rmtree(self.temporary_dir, ignore_errors=True)
         return True
@@ -300,65 +288,92 @@ class gatheringFiles:
             self.utils.request_input()
             return []
         
-    def gather_hardware_sniffer(self):
-        if os_name != "Windows":
-            return
-        
-        self.utils.head("Gathering Files")
-        print("")
-        print("Please wait for download Hardware Sniffer")
-        print("")
-
-        product_name = "Hardware-Sniffer-CLI.exe"
-
-        hardware_sniffer_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), product_name)
-
-        download_history = self.utils.read_file(self.download_history_file)
-
-        if not isinstance(download_history, list):
-            download_history = []
-
-        product_id = product_download_url = None 
-
-        latest_release = self.github.get_latest_release("lzhoang2801", "Hardware-Sniffer") or {}
-
-        for product in latest_release.get("assets"):
-            if product.get("product_name") == product_name.split(".")[0]:
-                _, product_id, product_download_url = product.values()
-        
+    def _update_download_history(self, download_history, product_name, product_id, product_url, sha256_hash):
         product_history_index = self.get_product_index(download_history, product_name)
         
-        print("")
-        if product_history_index == None:
-            print("Please wait for download {}...".format(product_name))
-        else:
-            if product_id == download_history[product_history_index].get("id") and os.path.exists(hardware_sniffer_path):
-                print("Latest version of {} already downloaded.".format(product_name))
-                return hardware_sniffer_path
-            else:
-                print("Updating {}...".format(product_name))
-
-        if product_download_url:
-            print("from " + product_download_url)
-        else:
-            print("Could not find download URL for {}.".format(product_name))
-            print("Please try again later.")
-            print("")
-            self.utils.request_input()
-            return
-        print("")
-
-        self.fetcher.download_and_save_file(product_download_url, hardware_sniffer_path)
+        entry = {
+            "product_name": product_name, 
+            "id": product_id,
+            "url": product_url,
+            "sha256": sha256_hash
+        }
 
         if product_history_index is None:
-            download_history.append({
-                "product_name": product_name, 
-                "id": product_id
-            })
+            download_history.append(entry)
         else:
-            download_history[product_history_index]["id"] = product_id
+            download_history[product_history_index].update(entry)
         
         self.utils.create_folder(os.path.dirname(self.download_history_file))
         self.utils.write_file(self.download_history_file, download_history)
         
-        return hardware_sniffer_path
+    def gather_hardware_sniffer(self):
+        if os_name != "Windows":
+            return
+
+        self.utils.head("Gathering Hardware Sniffer")
+
+        PRODUCT_NAME = "Hardware-Sniffer-CLI.exe"
+        REPO_OWNER = "lzhoang2801"
+        REPO_NAME = "Hardware-Sniffer"
+
+        destination_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), PRODUCT_NAME)
+        
+        latest_release = self.github.get_latest_release(REPO_OWNER, REPO_NAME) or {}
+        
+        product_id = None
+        product_download_url = None
+        sha256_hash = None
+
+        asset_name = PRODUCT_NAME.split('.')[0]
+        for asset in latest_release.get("assets", []):
+            if asset.get("product_name") == asset_name:
+                product_id = asset.get("id")
+                product_download_url = asset.get("url")
+                sha256_hash = asset.get("sha256")
+                break
+
+        if not all([product_id, product_download_url, sha256_hash]):
+            print("")
+            print("Could not find release information for {}.".format(PRODUCT_NAME))
+            print("Please try again later.")
+            print("")
+            self.utils.request_input()
+            raise Exception("Could not find release information for {}.".format(PRODUCT_NAME))
+
+        download_history = self.utils.read_file(self.download_history_file)
+        if not isinstance(download_history, list):
+            download_history = []
+
+        product_history_index = self.get_product_index(download_history, PRODUCT_NAME)
+        
+        if product_history_index is not None:
+            history_item = download_history[product_history_index]
+            is_latest_id = (product_id == history_item.get("id"))
+            
+            file_is_valid = False
+            if os.path.exists(destination_path):
+                local_hash = self.integrity_checker.get_sha256(destination_path)
+                file_is_valid = (sha256_hash == local_hash)
+
+            if is_latest_id and file_is_valid:
+                print("")
+                print("Latest version of {} already downloaded.".format(PRODUCT_NAME))
+                return destination_path
+
+        print("")
+        print("Updating" if product_history_index is not None else "Please wait for download", end=" ")
+        print("{}...".format(PRODUCT_NAME))
+        print("")
+        print("from {}".format(product_download_url))
+        print("")
+        
+        if not self.fetcher.download_and_save_file(product_download_url, destination_path, sha256_hash):
+            manual_download_url = f"https://github.com/{REPO_OWNER}/{REPO_NAME}/releases/latest"
+            print("Go to {} to download {} manually.".format(manual_download_url, PRODUCT_NAME))
+            print("")
+            self.utils.request_input()
+            raise Exception("Failed to download {}.".format(PRODUCT_NAME))
+
+        self._update_download_history(download_history, PRODUCT_NAME, product_id, product_download_url, sha256_hash)
+        
+        return destination_path

Scripts/github.py

@@ -82,32 +82,37 @@ class Github:
         assets = []
 
         in_li_block = False
-        download_link = None
 
         for line in response.splitlines():
 
             if "<li" in line:
                 in_li_block = True
-            elif "</li" in line:
-                in_li_block = False
                 download_link = None
+                sha256 = None
+                asset_id = None
+            elif in_li_block and "</li" in line:
+                if download_link and asset_id:
+                    assets.append({
+                        "product_name": self.extract_asset_name(download_link.split("/")[-1]), 
+                        "id": int(asset_id), 
+                        "url": "https://github.com" + download_link,
+                        "sha256": sha256
+                    })
+                in_li_block = False
 
             if in_li_block:  
-                if "<a" in line and "href=\"" in line and "/releases/download" in line:
+                if download_link is None and "<a" in line and "href=\"" in line and "/releases/download" in line:
                     download_link = line.split("href=\"", 1)[1].split("\"", 1)[0]
 
                     if not ("tlwm" in download_link or ("tlwm" not in download_link and "DEBUG" not in download_link.upper())):
                         in_li_block = False
-                        download_link = None
                         continue
 
-                if download_link and "<relative-time" in line:
+                if sha256 is None and "sha256:" in line:
+                    sha256 = line.split("sha256:", 1)[1].split("<", 1)[0]
+
+                if asset_id is None and "<relative-time" in line:
                     asset_id = self._generate_asset_id(line)
-                    assets.append({
-                        "product_name": self.extract_asset_name(download_link.split("/")[-1]), 
-                        "id": int(asset_id), 
-                        "url": "https://github.com" + download_link
-                    })
 
         return assets
 

Scripts/integrity_checker.py

@@ -0,0 +1,86 @@
+import os
+import hashlib
+import json
+from Scripts import utils
+
+class IntegrityChecker:
+    def __init__(self):
+        self.utils = utils.Utils()
+
+    def get_sha256(self, file_path, block_size=65536):
+        if not os.path.exists(file_path) or os.path.isdir(file_path):
+            return None
+
+        sha256 = hashlib.sha256()
+        with open(file_path, 'rb') as f:
+            for block in iter(lambda: f.read(block_size), b''):
+                sha256.update(block)
+        return sha256.hexdigest()
+
+    def generate_folder_manifest(self, folder_path, manifest_path=None):
+        if not os.path.isdir(folder_path):
+            return None
+
+        if manifest_path is None:
+            manifest_path = os.path.join(folder_path, "manifest.json")
+
+        manifest_data = {}
+        for root, _, files in os.walk(folder_path):
+            for name in files:
+                file_path = os.path.join(root, name)
+                relative_path = os.path.relpath(file_path, folder_path).replace('\\', '/')
+                
+                if relative_path == os.path.basename(manifest_path):
+                    continue
+
+                manifest_data[relative_path] = self.get_sha256(file_path)
+        
+        self.utils.write_file(manifest_path, manifest_data)
+        return manifest_data
+
+    def verify_folder_integrity(self, folder_path, manifest_path=None):
+        if not os.path.isdir(folder_path):
+            return None, "Folder not found."
+
+        if manifest_path is None:
+            manifest_path = os.path.join(folder_path, "manifest.json")
+
+        if not os.path.exists(manifest_path):
+            return None, "Manifest file not found."
+
+        manifest_data = self.utils.read_file(manifest_path)
+        if not isinstance(manifest_data, dict):
+            return None, "Invalid manifest file."
+            
+        issues = {
+            "modified": [],
+            "missing": [],
+            "untracked": []
+        }
+
+        manifest_files = set(manifest_data.keys())
+        actual_files = set()
+
+        for root, _, files in os.walk(folder_path):
+            for name in files:
+                file_path = os.path.join(root, name)
+                relative_path = os.path.relpath(file_path, folder_path)
+
+                if relative_path == os.path.basename(manifest_path):
+                    continue
+                
+                actual_files.add(relative_path)
+
+                if relative_path not in manifest_data:
+                    issues["untracked"].append(relative_path)
+                else:
+                    current_hash = self.get_sha256(file_path)
+                    if current_hash != manifest_data.get(relative_path):
+                        issues["modified"].append(relative_path)
+
+        missing_files = manifest_files - actual_files
+        issues["missing"] = list(missing_files)
+
+        is_valid = not any(issues.values())
+        
+        return is_valid, issues

Scripts/resource_fetcher.py

@@ -1,3 +1,5 @@
+from Scripts import integrity_checker
+from Scripts import utils
 import ssl
 import os
 import json
@@ -15,6 +17,8 @@ else:
     import urllib2
     from urllib2 import urlopen, Request, URLError
 
+MAX_ATTEMPTS = 3
+
 class ResourceFetcher:
     def __init__(self, headers=None):
         self.request_headers = headers or {
@@ -22,6 +26,8 @@ class ResourceFetcher:
         }
         self.buffer_size = 16 * 1024
         self.ssl_context = self.create_ssl_context()
+        self.integrity_checker = integrity_checker.IntegrityChecker()
+        self.utils = utils.Utils()
 
     def create_ssl_context(self):
         try:
@@ -149,22 +155,40 @@ class ResourceFetcher:
             
         print()
 
-    def download_and_save_file(self, resource_url, destination_path):
+    def download_and_save_file(self, resource_url, destination_path, sha256_hash=None):
         attempt = 0
 
-        while attempt < 3:
+        while attempt < MAX_ATTEMPTS:
+            attempt += 1
             response = self._make_request(resource_url)
 
             if not response:
-                attempt += 1
-                print("Failed to download file from {}. Retrying...".format(resource_url))
+                print("Failed to fetch content from {}. Retrying...".format(resource_url))
                 continue
 
-            self._download_with_progress(response, open(destination_path, "wb"))
+            with open(destination_path, "wb") as local_file:
+                self._download_with_progress(response, local_file)
 
             if os.path.exists(destination_path) and os.path.getsize(destination_path) > 0:
+                if sha256_hash:
+                    print("Verifying SHA256 checksum...")
+                    downloaded_hash = self.integrity_checker.get_sha256(destination_path)
+                    if downloaded_hash.lower() == sha256_hash.lower():
+                        print("Checksum verified successfully.")
+                        return True
+                    else:
+                        print("Checksum mismatch! Removing file and retrying download...")
+                        os.remove(destination_path)
+                        continue
+                else:
+                    print("No SHA256 hash provided. Downloading file without verification.")
                     return True
             
-            attempt += 1
+            if os.path.exists(destination_path):
+                os.remove(destination_path)
 
+            if attempt < MAX_ATTEMPTS:
+                print("Download failed for {}. Retrying...".format(resource_url))
+
+        print("Failed to download {} after {} attempts.".format(resource_url, MAX_ATTEMPTS))
         return False

Scripts/smbios.py

@@ -29,7 +29,7 @@ class SMBIOS:
             if download_history:
                 product_index = self.g.get_product_index(download_history, "OpenCorePkg")
                 
-                if product_index:
+                if product_index is not None:
                     download_history.pop(product_index)
                     self.utils.write_file(self.g.download_history_file, download_history)