feat(updater): add smooth splash GUI and active vanilla interceptor during updates

This commit is contained in:
tomasalias 2026-10-12 17:12:24 +02:00
parent a8f11dcd53
commit a4e88c980d
2 changed files with 456 additions and 45 deletions

View file

@ -22,6 +22,7 @@ RUN_KEY_PATH = r"Software\Microsoft\Windows\CurrentVersion\Run"
WATCHER_VALUE_NAME = "BetterCurseForgeWatcher"
WATCHER_MUTEX_NAME = "BetterCurseForgeWatcherMutex"
PATCH_MUTEX_NAME = "BetterCurseForgePatchMutex"
SPLASH_MUTEX_NAME = "BetterCurseForgeSplashMutex"
PERMANENT_DIR = Path(os.environ.get("LOCALAPPDATA", "")) / "BetterCurseForge"
PERMANENT_EXE = PERMANENT_DIR / "BetterCurseForge.exe"
PERMANENT_BCF = PERMANENT_DIR / "bettercf"
@ -112,13 +113,79 @@ def get_curseforge_dir() -> Path | None:
return p
return None
def get_process_pids(image_name: str) -> list[int]:
"""
Ultra-fast process lookup via Win32 Toolhelp32Snapshot (<1ms, zero subprocess overhead).
"""
if os.name != "nt":
return []
try:
import ctypes
from ctypes import wintypes
TH32CS_SNAPPROCESS = 0x00000002
class PROCESSENTRY32(ctypes.Structure):
_fields_ = [
('dwSize', wintypes.DWORD),
('cntUsage', wintypes.DWORD),
('th32ProcessID', wintypes.DWORD),
('th32DefaultHeapID', ctypes.c_size_t),
('th32ModuleID', wintypes.DWORD),
('cntThreads', wintypes.DWORD),
('th32ParentProcessID', wintypes.DWORD),
('pcPriClassBase', ctypes.c_long),
('dwFlags', wintypes.DWORD),
('szExeFile', ctypes.c_char * 260)
]
kernel32 = ctypes.windll.kernel32
h_snap = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
if h_snap == -1 or not h_snap:
return []
pe = PROCESSENTRY32()
pe.dwSize = ctypes.sizeof(PROCESSENTRY32)
pids = []
target = image_name.lower()
if kernel32.Process32First(h_snap, ctypes.byref(pe)):
while True:
name = pe.szExeFile.decode('mbcs', errors='ignore').lower()
if name == target:
pids.append(pe.th32ProcessID)
if not kernel32.Process32Next(h_snap, ctypes.byref(pe)):
break
kernel32.CloseHandle(h_snap)
return pids
except Exception:
return []
def kill_process_by_name(image_name: str) -> int:
"""
Silently and immediately terminates all matching processes via TerminateProcess (<1ms).
"""
if os.name != "nt":
return 0
try:
import ctypes
PROCESS_TERMINATE = 0x0001
kernel32 = ctypes.windll.kernel32
pids = get_process_pids(image_name)
killed = 0
for pid in pids:
h_proc = kernel32.OpenProcess(PROCESS_TERMINATE, False, pid)
if h_proc:
if kernel32.TerminateProcess(h_proc, 1):
killed += 1
kernel32.CloseHandle(h_proc)
return killed
except Exception:
return 0
def kill_curseforge():
kill_process_by_name("curseforge.exe")
try:
subprocess.run(
["taskkill", "/f", "/im", "CurseForge.exe"],
capture_output=True, timeout=10, creationflags=CREATE_NO_WINDOW
capture_output=True, timeout=5, creationflags=CREATE_NO_WINDOW
)
time.sleep(0.5)
time.sleep(0.3)
except Exception:
pass
@ -741,6 +808,8 @@ def unpatch_curseforge(log_callback=print) -> bool:
# ═══════════════════════════════════════════════════════════════════════════════
# AUTO-REPATCH WATCHER & PERSISTENCE ENGINE
def is_curseforge_running() -> bool:
if os.name == "nt":
return len(get_process_pids("curseforge.exe")) > 0
try:
out = subprocess.check_output(
["tasklist", "/FI", "IMAGENAME eq CurseForge.exe", "/NH"],
@ -751,6 +820,46 @@ def is_curseforge_running() -> bool:
return False
def is_updater_running() -> bool:
if os.name == "nt":
try:
import ctypes
from ctypes import wintypes
TH32CS_SNAPPROCESS = 0x00000002
class PROCESSENTRY32(ctypes.Structure):
_fields_ = [
('dwSize', wintypes.DWORD),
('cntUsage', wintypes.DWORD),
('th32ProcessID', wintypes.DWORD),
('th32DefaultHeapID', ctypes.c_size_t),
('th32ModuleID', wintypes.DWORD),
('cntThreads', wintypes.DWORD),
('th32ParentProcessID', wintypes.DWORD),
('pcPriClassBase', ctypes.c_long),
('dwFlags', wintypes.DWORD),
('szExeFile', ctypes.c_char * 260)
]
kernel32 = ctypes.windll.kernel32
h_snap = kernel32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0)
if h_snap != -1 and h_snap:
pe = PROCESSENTRY32()
pe.dwSize = ctypes.sizeof(PROCESSENTRY32)
found = False
if kernel32.Process32First(h_snap, ctypes.byref(pe)):
while True:
name = pe.szExeFile.decode('mbcs', errors='ignore').lower()
if name in ("setup.exe", "installer.exe", "update.exe") or \
("curseforge" in name and any(k in name for k in ("update", "setup", "install"))) or \
("overwolf" in name and any(k in name for k in ("update", "setup", "install"))):
found = True
break
if not kernel32.Process32Next(h_snap, ctypes.byref(pe)):
break
kernel32.CloseHandle(h_snap)
if found:
return True
except Exception:
pass
try:
out = subprocess.check_output(
["tasklist", "/FO", "CSV", "/NH"],
@ -769,6 +878,50 @@ def is_updater_running() -> bool:
pass
return False
def is_splash_running() -> bool:
if os.name != "nt":
return False
try:
import ctypes
kernel32 = ctypes.windll.kernel32
kernel32.OpenMutexW.restype = ctypes.c_void_p
kernel32.OpenMutexW.argtypes = [ctypes.c_uint, ctypes.c_bool, ctypes.c_wchar_p]
kernel32.CloseHandle.argtypes = [ctypes.c_void_p]
SYNCHRONIZE = 0x00100000
h = kernel32.OpenMutexW(SYNCHRONIZE, False, SPLASH_MUTEX_NAME)
if h:
kernel32.CloseHandle(h)
return True
return False
except Exception:
return False
def launch_splash_updater(pid_to_wait: int | None = None) -> bool:
if is_splash_running():
return False
perm_exe = ensure_permanent_installation()
args = ["--repatch-splash"]
if pid_to_wait:
args.append(str(pid_to_wait))
if perm_exe.exists() and perm_exe.suffix.lower() == ".exe":
cmd = [str(perm_exe.resolve())] + args
elif getattr(sys, "frozen", False):
cmd = [str(Path(sys.executable).resolve())] + args
else:
python_exe = Path(sys.executable).parent / "pythonw.exe"
if not python_exe.exists():
python_exe = Path(sys.executable)
cmd = [str(python_exe.resolve()), str(Path(__file__).parent.resolve() / "patcher_gui.py")] + args
try:
creation_flags = 0x00000008 | 0x00000200 if os.name == "nt" else 0 # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
subprocess.Popen(cmd, creationflags=creation_flags)
return True
except Exception:
return False
def ensure_permanent_installation() -> Path:
"""
Ensures BetterCurseForge executable, bettercf runtime, and vendor tools are permanently
@ -909,10 +1062,12 @@ def stop_background_watcher():
except Exception:
pass
def wait_for_files_settled(cf_dir: Path, timeout: float = 90.0, log_fn=None) -> bool:
def wait_for_files_settled(cf_dir: Path, timeout: float = 90.0, log_fn=None, intercept_unpatched: bool = True) -> bool:
"""
Waits for the CurseForge updater/installer to finish writing files.
Ensures resources/app.asar is present, fully written, and readable.
Actively intercepts and kills any unpatched CurseForge.exe instances launched
by the installer before their window can render on screen.
"""
import time
start = time.time()
@ -921,7 +1076,15 @@ def wait_for_files_settled(cf_dir: Path, timeout: float = 90.0, log_fn=None) ->
stable_checks = 0
while time.time() - start < timeout:
time.sleep(1.5)
# Check in 150ms slices (total ~1.5s per outer cycle) to actively silence unpatched CurseForge
for _ in range(10):
if intercept_unpatched:
pids = get_process_pids("curseforge.exe")
if pids:
kill_process_by_name("curseforge.exe")
if log_fn:
log_fn("Silencing vanilla launcher in background...")
time.sleep(0.15)
if is_updater_running():
stable_checks = 0
@ -958,9 +1121,12 @@ def wait_for_files_settled(cf_dir: Path, timeout: float = 90.0, log_fn=None) ->
def wait_for_update_and_repatch(pid_to_wait: int | None = None, log_fn=None) -> bool:
"""
Invoked when CurseForge is quitting to apply an update.
Waits for the old CurseForge instance to exit, waits for the installer to finish,
and immediately repatches before/as CurseForge restarts.
Spawns the dedicated Splash Updater GUI for a smooth visual transition,
or falls back to headless repatch if UI cannot launch.
"""
if launch_splash_updater(pid_to_wait):
return True
import time
if log_fn is None:
log_dir = get_user_data_dir()
@ -983,27 +1149,19 @@ def wait_for_update_and_repatch(pid_to_wait: int | None = None, log_fn=None) ->
log_fn(f"Waiting for PID {pid_to_wait} to terminate...")
for _ in range(30):
try:
out = subprocess.check_output(
["tasklist", "/FI", f"PID eq {pid_to_wait}", "/NH"],
text=True, stderr=subprocess.DEVNULL, creationflags=CREATE_NO_WINDOW
)
if str(pid_to_wait) not in out:
pids = get_process_pids("curseforge.exe")
if pid_to_wait not in pids:
break
except Exception:
break
time.sleep(0.5)
settled = wait_for_files_settled(cf_dir, timeout=90.0, log_fn=log_fn)
settled = wait_for_files_settled(cf_dir, timeout=90.0, log_fn=log_fn, intercept_unpatched=True)
if not settled:
log_fn("Timed out waiting for installer to settle.")
return False
was_running = is_curseforge_running()
if was_running:
log_fn("CurseForge launched unpatched by installer. Terminating to apply patch...")
kill_curseforge()
time.sleep(1.5)
kill_curseforge()
success = False
for attempt in range(1, 4):
success = patch_curseforge(log_callback=log_fn)
@ -1058,35 +1216,39 @@ def run_watcher_loop():
wlog("CurseForge directory not found. Watcher exiting.")
return
last_splash_launch = 0.0
def check_and_repatch(trigger_reason: str):
if not is_curseforge_patched(cf_dir):
wlog(f"CurseForge unpatched state detected ({trigger_reason}). Waiting for files to settle...")
settled = wait_for_files_settled(cf_dir, timeout=90.0, log_fn=wlog)
if settled:
was_running = is_curseforge_running()
if was_running:
wlog("CurseForge is currently running unpatched. Closing it to apply BetterCurseForge patch...")
nonlocal last_splash_launch
now = time.time()
if is_splash_running():
return
needs_patch = not is_curseforge_patched(cf_dir) or is_updater_running()
if needs_patch:
if now - last_splash_launch < 12.0:
return
last_splash_launch = now
wlog(f"CurseForge update or unpatched state detected ({trigger_reason}). Launching Splash Updater...")
ok = launch_splash_updater()
if not ok:
wlog("Could not launch Splash Updater; falling back to background patch.")
settled = wait_for_files_settled(cf_dir, timeout=90.0, log_fn=wlog, intercept_unpatched=True)
if settled:
kill_curseforge()
time.sleep(1.5)
wlog("Applying BetterCurseForge patch...")
success = False
for attempt in range(1, 4):
success = patch_curseforge(log_callback=wlog)
success = False
for attempt in range(1, 4):
success = patch_curseforge(log_callback=wlog)
if success:
break
time.sleep(2)
if success:
break
wlog(f"Patch attempt {attempt} failed, retrying in 2 seconds...")
time.sleep(2)
if success:
wlog("★ CurseForge successfully re-patched after update!")
if was_running:
wlog("Relaunching modded CurseForge...")
wlog("★ CurseForge successfully re-patched after update!")
subprocess.Popen([str(cf_dir / "CurseForge.exe")], creationflags=CREATE_NO_WINDOW)
else:
wlog("Failed to re-patch CurseForge after update.")
else:
wlog("Failed to re-patch CurseForge after update.")
else:
wlog("Timed out waiting for installer files to settle.")
wlog("Timed out waiting for installer files to settle.")
check_and_repatch("Startup Check")

View file

@ -9,7 +9,7 @@ and providing zero-timer automatic re-patching after updates.
import os
import sys
if os.name == "nt":
if len(sys.argv) > 1 and sys.argv[1].lower() not in ("--watch", "--wait-for-update", "--ensure-watcher"):
if len(sys.argv) > 1 and sys.argv[1].lower() not in ("--watch", "--wait-for-update", "--ensure-watcher", "--repatch-splash", "--splash"):
import ctypes
ctypes.windll.kernel32.AttachConsole(-1)
if sys.stdout is not None:
@ -469,6 +469,249 @@ class BetterCFApp:
self.log(f"BetterCurseForge Patcher v{VERSION} ready.")
self.root.mainloop()
# ═══════════════════════════════════════════════════════════════════════════════
# DEDICATED SPLASH UPDATER GUI
# ═══════════════════════════════════════════════════════════════════════════════
class SplashUpdaterWindow:
def __init__(self, target_pid: int | None = None):
self.target_pid = target_pid
self.root = tk.Tk()
self.root.title("BetterCurseForge Updater")
width, height = 460, 215
screen_w = self.root.winfo_screenwidth()
screen_h = self.root.winfo_screenheight()
x = (screen_w - width) // 2
y = (screen_h - height) // 2
self.root.geometry(f"{width}x{height}+{x}+{y}")
self.root.overrideredirect(True)
self.root.attributes("-topmost", True)
self.root.configure(bg=COLORS["bg"])
self._icon_photo = None
self._drag_x = 0
self._drag_y = 0
self._setup_window_icon()
self._build_ui()
def _setup_window_icon(self):
ico_path = find_asset("icon.ico")
if ico_path and os.name == "nt":
try:
self.root.iconbitmap(str(ico_path.resolve()))
self.root.iconbitmap(default=str(ico_path.resolve()))
except Exception:
pass
def _build_ui(self):
card = tk.Frame(
self.root, bg=COLORS["bg_card"],
highlightbackground=COLORS["purple_dark"], highlightthickness=1,
padx=20, pady=18
)
card.pack(fill="both", expand=True)
def start_drag(e):
self._drag_x = e.x
self._drag_y = e.y
def do_drag(e):
x = self.root.winfo_x() + (e.x - self._drag_x)
y = self.root.winfo_y() + (e.y - self._drag_y)
self.root.geometry(f"+{x}+{y}")
card.bind("<Button-1>", start_drag)
card.bind("<B1-Motion>", do_drag)
# Header
header = tk.Frame(card, bg=COLORS["bg_card"])
header.pack(fill="x")
header.bind("<Button-1>", start_drag)
header.bind("<B1-Motion>", do_drag)
png_path = find_asset("icon_512.png") or find_asset("icon.png")
if png_path and HAS_PIL:
try:
im = Image.open(png_path).resize((44, 44), Image.Resampling.LANCZOS)
self._icon_photo = ImageTk.PhotoImage(im)
icon_lbl = tk.Label(header, image=self._icon_photo, bg=COLORS["bg_card"])
icon_lbl.pack(side="left", padx=(0, 14))
icon_lbl.bind("<Button-1>", start_drag)
icon_lbl.bind("<B1-Motion>", do_drag)
except Exception:
pass
text_col = tk.Frame(header, bg=COLORS["bg_card"])
text_col.pack(side="left", fill="both", expand=True)
text_col.bind("<Button-1>", start_drag)
text_col.bind("<B1-Motion>", do_drag)
title_row = tk.Frame(text_col, bg=COLORS["bg_card"])
title_row.pack(anchor="w")
title_row.bind("<Button-1>", start_drag)
title_row.bind("<B1-Motion>", do_drag)
t_lbl = tk.Label(
title_row, text="BetterCurseForge", font=("Segoe UI", 13, "bold"),
fg=COLORS["text"], bg=COLORS["bg_card"]
)
t_lbl.pack(side="left")
t_lbl.bind("<Button-1>", start_drag)
t_lbl.bind("<B1-Motion>", do_drag)
badge = tk.Frame(
title_row, bg="#261b3d", highlightbackground=COLORS["purple_glow"],
highlightthickness=1, padx=6, pady=1
)
badge.pack(side="left", padx=(10, 0))
b_lbl = tk.Label(
badge, text="AUTO-UPDATE", font=("Segoe UI", 8, "bold"),
fg="#c4b5fd", bg="#261b3d"
)
b_lbl.pack()
sub_lbl = tk.Label(
text_col, text="CurseForge update detected. Applying enhancements...",
font=("Segoe UI", 9), fg=COLORS["text_dim"], bg=COLORS["bg_card"]
)
sub_lbl.pack(anchor="w", pady=(3, 0))
sub_lbl.bind("<Button-1>", start_drag)
sub_lbl.bind("<B1-Motion>", do_drag)
# Progress bar
style = ttk.Style()
style.theme_use("clam")
style.configure(
"Splash.Horizontal.TProgressbar",
troughcolor="#1e1f22",
background=COLORS["purple"],
bordercolor="#1e1f22",
lightcolor=COLORS["purple"],
darkcolor=COLORS["purple"]
)
self.pbar = ttk.Progressbar(
card, style="Splash.Horizontal.TProgressbar",
mode="indeterminate", length=410
)
self.pbar.pack(fill="x", pady=(16, 10))
self.pbar.start(12)
# Status labels
self.status_lbl = tk.Label(
card, text="Waiting for CurseForge installer to finish...",
font=("Segoe UI", 9, "bold"), fg=COLORS["text"], bg=COLORS["bg_card"], anchor="w"
)
self.status_lbl.pack(fill="x")
self.detail_lbl = tk.Label(
card, text="Silencing vanilla client in background...",
font=("Segoe UI", 8), fg=COLORS["text_muted"], bg=COLORS["bg_card"], anchor="w"
)
self.detail_lbl.pack(fill="x", pady=(2, 0))
def set_status(self, main_text: str, detail_text: str = "", status_type: str = "normal"):
color = COLORS["text"]
if status_type == "success":
color = COLORS["green"]
elif status_type == "error":
color = COLORS["red"]
elif status_type == "warn":
color = COLORS["yellow"]
self.status_lbl.configure(text=main_text, fg=color)
if detail_text:
self.detail_lbl.configure(text=detail_text)
def run(self):
threading.Thread(target=self._worker, daemon=True).start()
self.root.mainloop()
def _worker(self):
import time
with patch_core.GlobalNamedMutex(patch_core.SPLASH_MUTEX_NAME, timeout_ms=500) as splash_lock:
if not splash_lock.acquired:
self.root.after(0, self.root.destroy)
return
try:
cf_dir = patch_core.get_curseforge_dir()
if not cf_dir:
self.root.after(0, self.set_status, "CurseForge not found!", "Please reinstall CurseForge.", "error")
time.sleep(3.0)
return
if self.target_pid:
self.root.after(0, self.set_status, "Waiting for previous instance to exit...", f"CurseForge PID {self.target_pid}")
for _ in range(30):
pids = patch_core.get_process_pids("curseforge.exe")
if self.target_pid not in pids:
break
time.sleep(0.5)
self.root.after(0, self.set_status, "Waiting for CurseForge installer...", "Silencing vanilla launcher in background...")
def on_settle_log(msg: str):
if "running" in msg.lower():
self.root.after(0, self.set_status, "CurseForge installer running...", "Silencing vanilla launcher in background...")
elif "settle" in msg.lower() or "mb" in msg.lower():
self.root.after(0, self.set_status, "Finalizing update files...", msg)
elif "silenc" in msg.lower():
self.root.after(0, self.set_status, "Silenced vanilla launcher in background...", "Preventing unpatched window display...")
settled = patch_core.wait_for_files_settled(cf_dir, timeout=90.0, log_fn=on_settle_log, intercept_unpatched=True)
if not settled:
self.root.after(0, self.set_status, "Installer wait timed out.", "Will attempt patch now...", "warn")
patch_core.kill_curseforge()
self.root.after(0, self.set_status, "Extracting application archive...", "Unpacking update files...")
def on_patch_log(msg: str):
if "Extracting" in msg:
self.root.after(0, self.set_status, "Extracting application archive...", "Unpacking update files...")
elif "Unlocked native premium" in msg:
self.root.after(0, self.set_status, "Unlocking native premium themes...", "Patching client theme routes...")
elif "premium theme state" in msg or "subscription plan" in msg:
self.root.after(0, self.set_status, "Enabling theme engine...", "Bypassing subscription restrictions...")
elif "Injected zero-ad" in msg:
self.root.after(0, self.set_status, "Injecting ad-block rules...", "Removing sponsored banners and ad containers...")
elif "Bundled BetterCurseForge runtime" in msg:
self.root.after(0, self.set_status, "Bundling BetterCurseForge runtime...", "Injecting core modules...")
elif "main-process hooks" in msg:
self.root.after(0, self.set_status, "Hooking main process...", "Configuring network filter and IPC...")
elif "Repacking app.asar" in msg:
self.root.after(0, self.set_status, "Repacking application archive...", "Rebuilding app.asar package...")
elif "integrity signature" in msg:
self.root.after(0, self.set_status, "Updating security signatures...", "Synchronizing OWEASARSIG header hash...")
elif "successfully patched" in msg:
self.root.after(0, self.set_status, "✓ Complete! Launching BetterCurseForge...", "All enhancements applied successfully.", "success")
success = False
for attempt in range(1, 4):
success = patch_core.patch_curseforge(log_callback=on_patch_log)
if success:
break
self.root.after(0, self.set_status, f"Retrying patch ({attempt+1}/3)...", "Resolving temporary file lock...", "warn")
time.sleep(2.0)
if success:
self.root.after(0, self.set_status, "✓ Complete! Launching BetterCurseForge...", "Enjoy your ad-free experience!", "success")
exe = cf_dir / "CurseForge.exe"
if exe.exists():
creation_flags = 0x00000008 | 0x00000200 if os.name == "nt" else 0 # DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP
subprocess.Popen([str(exe)], cwd=str(cf_dir), creationflags=creation_flags)
time.sleep(0.9)
else:
self.root.after(0, self.set_status, "Failed to apply patch.", "Check watcher.log for details.", "error")
time.sleep(3.0)
except Exception as e:
self.root.after(0, self.set_status, "Error during update.", str(e), "error")
time.sleep(3.0)
finally:
self.root.after(0, self.root.destroy)
# ═══════════════════════════════════════════════════════════════════════════════
# MAIN ENTRY POINT
# ═══════════════════════════════════════════════════════════════════════════════
@ -480,9 +723,15 @@ if __name__ == "__main__":
if arg == "--watch":
patch_core.run_watcher_loop()
sys.exit(0)
elif arg in ("--repatch-splash", "--splash"):
pid = int(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2].isdigit() else None
splash = SplashUpdaterWindow(pid)
splash.run()
sys.exit(0)
elif arg == "--wait-for-update":
pid = int(sys.argv[2]) if len(sys.argv) > 2 and sys.argv[2].isdigit() else None
patch_core.wait_for_update_and_repatch(pid)
splash = SplashUpdaterWindow(pid)
splash.run()
sys.exit(0)
elif arg == "--ensure-watcher":
if not patch_core.is_watcher_running():
@ -506,7 +755,7 @@ if __name__ == "__main__":
except SystemExit:
pass
except Exception as e:
if len(sys.argv) > 1 and sys.argv[1].lower() in ("--watch", "--wait-for-update", "--ensure-watcher"):
if len(sys.argv) > 1 and sys.argv[1].lower() in ("--watch", "--wait-for-update", "--ensure-watcher", "--repatch-splash", "--splash"):
try:
log_dir = patch_core.get_user_data_dir()
with open(log_dir / "watcher.log", "a", encoding="utf-8") as f: