from PYAS_Diagnostics import log_exception, bind_report_logging import os import sys import time import copy import json import uuid import queue import platform import threading import logging from PYAS_Startup import ( bootstrap_startup, check_webview_dependencies, prepare_webview_profile, restart_with_new_webview_profile, wait_for_recovery_parent, ) STARTUP_LOG_PATH = bootstrap_startup() if __name__ == "__main__" else None import msvcrt import winreg import pystray import subprocess import webview import webbrowser import ctypes import ctypes.wintypes from concurrent.futures import ThreadPoolExecutor from http.server import SimpleHTTPRequestHandler from PIL import Image from socketserver import ThreadingTCPServer from webview.dom import DOMEventHandler from PYAS_Engine import sign_scanner, rule_scanner, pe_scanner, cloud_scanner from PYAS_Protect import ProtectMixin from PYAS_Scanner import ScannerMixin from PYAS_Tools import ( COPYDATASTRUCT, FILE_NOTIFY_INFORMATION, FILTER_MESSAGE_HEADER, IO_COUNTERS, LUID, LUID_AND_ATTRIBUTES, MEMORY_BASIC_INFORMATION, MIB_TCPROW_OWNER_PID, POINT, PROCESSENTRY32W, PROCESS_BASIC_INFORMATION, PYAS_FULL_MESSAGE, PYAS_MESSAGE, PYAS_USER_MESSAGE, RECT, SERVICE_STATUS_PROCESS, SHQUERYRBINFO, TOKEN_PRIVILEGES, ToolsMixin, UNICODE_STRING, ) PYAS_WINDOW_TITLE = "PYAS Security" WM_COPYDATA = 0x004A SMTO_ABORTIFHUNG = 0x0002 SMTO_ERRORONEXIT = 0x0020 PYAS_SEND_MESSAGE_FLAGS = SMTO_ABORTIFHUNG | SMTO_ERRORONEXIT PYAS_MESSAGE_SCAN = 1 PYAS_MESSAGE_SHOW = 2 PYAS_MESSAGE_CLOSE = 3 PYAS_MESSAGE_QUIT = 4 PYAS_MESSAGE_DRIVER_UNLOAD = 5 PYAS_MESSAGE_DRIVER_UNINSTALL = 6 PYAS_MAINTENANCE_MESSAGE = 0x8000 + 0x501 MSGFLT_REMOVE = 2 MSGFLT_DISALLOW = 2 PYAS_MAINTENANCE_COMMANDS = frozenset( { PYAS_MESSAGE_CLOSE, PYAS_MESSAGE_QUIT, PYAS_MESSAGE_DRIVER_UNLOAD, PYAS_MESSAGE_DRIVER_UNINSTALL, } ) from PYAS_WinAPI import WindowsMixin from PYAS_Config import ConfigMixin from PYAS_Logs import LogMixin from PYAS_Runtime import RuntimeMixin class _MainMixin(WindowsMixin, ConfigMixin, LogMixin, RuntimeMixin): def __init__(self): self._window = None self.tray_icon = None self.logs_data = [] self.logs_dirty = False self.lock_config = threading.RLock() self.lock_logs = threading.RLock() self.lock_virus = threading.RLock() self.lock_file_ops = threading.RLock() self.init_environ() self.init_windll() if not self.check_singleton("PYAS_Security_Mutex"): forwarded_exit_code = self.forward_to_existing_instance() if forwarded_exit_code is not None: os._exit(forwarded_exit_code) self.h_recovery_mutex = self.kernel32.CreateMutexW( None, False, "PYAS_Security_Recovery_Mutex" ) if ctypes.get_last_error() == 183: os._exit(2) self.init_variables() self.load_config() self.load_logs() bind_report_logging(self) def find_existing_window(self, timeout=2.0): deadline = time.monotonic() + timeout while time.monotonic() < deadline: hwnd = self.user32.FindWindowW(None, PYAS_WINDOW_TITLE) if hwnd: return hwnd time.sleep(0.05) return None def send_existing_window_message(self, hwnd, message_id, payload=None, timeout=1000): if not hwnd: return False if message_id in PYAS_MAINTENANCE_COMMANDS: try: return bool( self.user32.SendMessageTimeoutW( hwnd, PYAS_MAINTENANCE_MESSAGE, message_id, None, PYAS_SEND_MESSAGE_FLAGS, timeout, None, ) ) except Exception: log_exception("MainMixin.send_maintenance_message") return False cds = COPYDATASTRUCT() cds.dwData = message_id buffer = None if payload is not None: encoded = str(payload).encode("utf-8") buffer = ctypes.create_string_buffer(encoded + b"\x00") cds.cbData = len(encoded) + 1 cds.lpData = ctypes.cast(buffer, ctypes.c_void_p) else: cds.cbData = 0 cds.lpData = None try: return bool( self.user32.SendMessageTimeoutW( hwnd, WM_COPYDATA, 0, ctypes.byref(cds), PYAS_SEND_MESSAGE_FLAGS, timeout, None ) ) except Exception: log_exception("PYAS._MainMixin.send_existing_window_message:96") return False def _is_driver_service_absent(self): scm = self.advapi32.OpenSCManagerW(None, None, 0x0001) if not scm: return False service = None try: ctypes.set_last_error(0) service = self.advapi32.OpenServiceW(scm, "PYAS_Driver", 0x0004) if service: return False return ctypes.get_last_error() == 1060 finally: if service: self.advapi32.CloseServiceHandle(service) self.advapi32.CloseServiceHandle(scm) def wait_for_existing_shutdown(self, timeout=30.0, require_service_removal=True): deadline = time.monotonic() + timeout while time.monotonic() < deadline: window_closed = not self.user32.FindWindowW(None, PYAS_WINDOW_TITLE) service_removed = not require_service_removal or self._is_driver_service_absent() if window_closed and service_removed: return True time.sleep(0.1) return False def forward_to_existing_instance(self): maintenance_request = any( arg in self.args_pyas for arg in ("-quit", "-driver-uninstall", "-driver-unload") ) hwnd = self.find_existing_window(timeout=10.0 if maintenance_request else 2.0) if not hwnd: return 2 if maintenance_request else None if "-quit" in self.args_pyas: if not self.send_existing_window_message(hwnd, PYAS_MESSAGE_QUIT, timeout=5000): return 2 return 0 if self.wait_for_existing_shutdown() else 2 if "-driver-uninstall" in self.args_pyas: if not self.send_existing_window_message( hwnd, PYAS_MESSAGE_DRIVER_UNINSTALL, timeout=5000 ): return 2 return 0 if self.wait_for_existing_shutdown() else 2 if "-driver-unload" in self.args_pyas: if not self.send_existing_window_message( hwnd, PYAS_MESSAGE_DRIVER_UNLOAD, timeout=5000 ): return 2 return 0 if self.wait_for_existing_shutdown(require_service_removal=False) else 2 if "-scan" in self.args_pyas: try: idx = self.args_pyas.index("-scan") target = self.args_pyas[idx + 1] except Exception: log_exception("PYAS._MainMixin.forward_to_existing_instance:155") return 2 if self.send_existing_window_message(hwnd, PYAS_MESSAGE_SCAN, target, timeout=1200): try: self.user32.SetForegroundWindow(hwnd) except Exception: log_exception("PYAS._MainMixin.forward_to_existing_instance:161") pass return 0 return 2 if self.send_existing_window_message(hwnd, PYAS_MESSAGE_SHOW, timeout=1200): try: self.user32.SetForegroundWindow(hwnd) except Exception: log_exception("PYAS._MainMixin.forward_to_existing_instance:169") pass return 0 return 2 def check_singleton(self, name): try: self.h_mutex = self.kernel32.CreateMutexW(None, False, name) if ctypes.get_last_error() == 183: return False return True except Exception: log_exception("PYAS._MainMixin.check_singleton:181") return False def get_tray_text(self, key): texts = { "open_ui": { "traditional_switch": "開啟介面", "simplified_switch": "打开界面", "english_switch": "Open PYAS", "japanese_switch": "PYAS を開く", "korean_switch": "PYAS 열기", "french_switch": "Ouvrir PYAS", "spanish_switch": "Abrir PYAS", "hindi_switch": "PYAS खोलें", "arabic_switch": "فتح PYAS", "russian_switch": "Открыть PYAS", "slovenian_switch": "Odpri PYAS", }, "optimize_mem": { "traditional_switch": "一鍵加速", "simplified_switch": "一键加速", "english_switch": "Memory Boost", "japanese_switch": "メモリ最適化", "korean_switch": "메모리 최적화", "french_switch": "Optimiser", "spanish_switch": "Optimizar", "hindi_switch": "मेमोरी बूस्ट", "arabic_switch": "تسريع", "russian_switch": "Ускорение", "slovenian_switch": "Optimizacija", }, "check_update": { "traditional_switch": "檢查更新", "simplified_switch": "检查更新", "english_switch": "Check Update", "japanese_switch": "更新を確認", "korean_switch": "업데이트 확인", "french_switch": "Vérifier la mise à jour", "spanish_switch": "Buscar actualizaciones", "hindi_switch": "अद्यतन जाँचे", "arabic_switch": "التحقق من التحديثات", "russian_switch": "Проверить обновления", "slovenian_switch": "Preveri posodobitve", }, "exit_app": { "traditional_switch": "退出防護", "simplified_switch": "退出防护", "english_switch": "Exit Security", "japanese_switch": "保護を終了", "korean_switch": "보호 종료", "french_switch": "Quitter la sécurité", "spanish_switch": "Salir de la seguridad", "hindi_switch": "सुरक्षा से बाहर निकलें", "arabic_switch": "خروج من الحماية", "russian_switch": "Выйти из защиты", "slovenian_switch": "Izhod iz zaščite", }, } return self._loc(texts.get(key, {})) def get_app_icon(self): icon_path = os.path.join(self.path_pyas, "Interface", "static", "img", "icon.ico") if os.path.exists(icon_path): try: return Image.open(icon_path) except Exception: log_exception("PYAS._MainMixin.get_app_icon:218") pass def show_tray(self): if self.tray_icon is not None: return menu = pystray.Menu( pystray.MenuItem( lambda item: self.get_tray_text("open_ui"), self.restore_from_tray, default=True ), pystray.MenuItem(lambda item: self.get_tray_text("optimize_mem"), self.optimize_memory), pystray.MenuItem( lambda item: self.get_tray_text("check_update"), self.tray_check_update ), pystray.MenuItem(lambda item: self.get_tray_text("exit_app"), self.close), ) self.tray_icon = pystray.Icon("PYAS", self.get_app_icon(), "PYAS Security", menu) self.tray_icon.run_detached() def tray_check_update(self, icon=None, item=None): def _check(): res = self.check_update() title_error = self._loc( { "traditional_switch": "錯誤", "simplified_switch": "错误", "english_switch": "Error", "japanese_switch": "エラー", "korean_switch": "오류", "french_switch": "Erreur", "spanish_switch": "Error", "hindi_switch": "त्रुटि", "arabic_switch": "خطأ", "russian_switch": "Ошибка", "slovenian_switch": "Napaka", } ) title_prompt = self._loc( { "traditional_switch": "提示", "simplified_switch": "提示", "english_switch": "Prompt", "japanese_switch": "プロンプト", "korean_switch": "프롬프트", "french_switch": "Indication", "spanish_switch": "Aviso", "hindi_switch": "सुझाव", "arabic_switch": "تلميح", "russian_switch": "Подсказка", "slovenian_switch": "Namig", } ) msg_fail = self._loc( { "traditional_switch": "檢查更新失敗", "simplified_switch": "检查更新失败", "english_switch": "Update check failed", "japanese_switch": "アップデートの確認に失敗しました", "korean_switch": "업데이트 확인 실패", "french_switch": "Échec de la vérification des mises à jour", "spanish_switch": "Fallo al buscar actualizaciones", "hindi_switch": "अपडेट की जाँच विफल रही", "arabic_switch": "فشل التحقق من التحديثات", "russian_switch": "Ошибка проверки обновлений", "slovenian_switch": "Preverjanje posodobitev ni uspelo", } ) msg_new = self._loc( { "traditional_switch": "發現新版本", "simplified_switch": "发现新版本", "english_switch": "New version found", "japanese_switch": "新しいバージョンが見つかりました", "korean_switch": "새 버전을 찾았습니다", "french_switch": "Nouvelle version trouvée", "spanish_switch": "Nueva versión encontrada", "hindi_switch": "नया संस्करण मिला", "arabic_switch": "تم العثور على إصدار جديد", "russian_switch": "Найдена новая версия", "slovenian_switch": "Najdena nova različica", } ) msg_latest = self._loc( { "traditional_switch": "當前已是最新版本", "simplified_switch": "当前已是最新版本", "english_switch": "Currently at latest version", "japanese_switch": "現在は最新バージョンです", "korean_switch": "현재 최신 버전입니다", "french_switch": "Actuellement à la dernière version", "spanish_switch": "Actualmente en la última versión", "hindi_switch": "वर्तमान में नवीनतम संस्करण है", "arabic_switch": "أنت تستخدم أحدث إصدار حاليًا", "russian_switch": "Установлена последняя версия", "slovenian_switch": "Trenutno imate najnovejšo različico", } ) if res.get("error"): self.show_alert(title_error, msg_fail, "error") elif res.get("has_update"): msg = ( f"{msg_new} {res.get('latest')}\n({res.get('current')} -> {res.get('latest')})" ) if self.show_confirm(title_prompt, msg): self.open_url(res.get("url")) else: msg = f"{msg_latest} {res.get('current')}" self.show_alert(title_prompt, msg, "info") threading.Thread(target=_check, daemon=True).start() def restore_from_tray(self, icon=None, item=None): if self._window: self._window.restore() self._window.show() hwnd = self.user32.FindWindowW(None, PYAS_WINDOW_TITLE) if hwnd: self.user32.SetForegroundWindow(hwnd) def set_window(self, window): self._window = window def minimize(self): if self._window: self._window.minimize() else: hwnd = self.user32.FindWindowW(None, PYAS_WINDOW_TITLE) if hwnd: self.user32.ShowWindow(hwnd, 6) def hide_window(self): if self._window: self._window.hide() def destroy_window_with_timeout(self, window, timeout=1.0): if not window: return done = threading.Event() def destroy_window(): try: window.destroy() except Exception: log_exception("PYAS._MainMixin.destroy_window_with_timeout.destroy_window:315") pass finally: done.set() threading.Thread(target=destroy_window, daemon=True).start() done.wait(timeout) def close(self, *args, uninstall_driver=False, **kwargs): if getattr(self, "closing", False): return True self.closing = True with self.lock_config: driver_enabled = self.pyas_config.get("driver_switch", False) driver_loaded = driver_enabled or self.check_system_driver() if uninstall_driver: driver_stopped, driver_error = self.uninstall_system_driver() elif driver_loaded: driver_stopped = self.stop_system_driver() driver_error = 0 if driver_stopped else 1051 else: driver_stopped, driver_error = True, 0 if not driver_stopped: self.closing = False action = "uninstall" if uninstall_driver else "unload" self.write_log( "WARN", "Driver Protection", detail=f"Controlled {action} failed: 0x{driver_error & 0xFFFFFFFF:08X}", success=False, ) if self._window: try: self._window.show() except Exception: log_exception("PYAS._MainMixin.close:348") pass return False if self.tray_icon: try: self.tray_icon.stop() except Exception: log_exception("PYAS._MainMixin.close:355") pass self.tray_icon = None window = self._window self._window = None if window: try: window.hide() except Exception: log_exception("PYAS._MainMixin.close:364") pass with self.lock_config: self.pyas_config["process_switch"] = False self.pyas_config["document_switch"] = False self.pyas_config["system_switch"] = False self.pyas_config["driver_switch"] = False self.pyas_config["network_switch"] = False self._cancel_pending_file_tasks() scheduler = getattr(self, "file_scheduler", None) if scheduler: scheduler.close() self.cloud_cancel_event.set() with self.lock_file_ops: if getattr(self, "h_dir_file", None): try: self.kernel32.CloseHandle(self.h_dir_file) except Exception: log_exception("PYAS._MainMixin.close:380") pass self.h_dir_file = None if hasattr(self, "virus_lock"): for file_path, (fd, lock_size) in list(self.virus_lock.items()): try: msvcrt.locking(fd, msvcrt.LK_UNLCK, lock_size) except Exception: log_exception("PYAS._MainMixin.close:388") pass try: os.close(fd) except Exception: log_exception("PYAS._MainMixin.close:392") pass self.virus_lock.clear() with self.lock_proc: if hasattr(self, "suspended_procs"): for h in list(self.suspended_procs): try: self.ntdll.NtResumeProcess(h) self.kernel32.CloseHandle(h) except Exception: log_exception("PYAS._MainMixin.close:402") pass self.suspended_procs.clear() for mutex_name in ("h_mutex", "h_recovery_mutex"): handle = getattr(self, mutex_name, None) if handle: try: self.kernel32.CloseHandle(handle) except Exception: log_exception("PYAS._MainMixin.close:411") pass setattr(self, mutex_name, None) self.flush_logs_now() self.destroy_window_with_timeout(window) os._exit(0) def report_ui_error(self, stage, message): logging.getLogger("PYAS.Startup").warning( "UI stage=%s: %s", str(stage)[:80], str(message)[:2000] ) return True def init_ui_ready(self): logging.getLogger("PYAS.Startup").info("Python UI bridge ready") self.ui_ready_event.set() with self.lock_config: if self.engine_initialized: return self.engine_initialized = True self.start_daemon_thread(self.init_engine_thread) def trigger_block_notification(self, action, source, target, code): if action not in [ "Process Block", "Process DLL Block", "File Block", "Network Block", "Driver Block", ]: return titles = { "Process Block": { "traditional_switch": "進程防護", "simplified_switch": "进程防护", "english_switch": "Process Protection", "japanese_switch": "プロセス保護", "korean_switch": "프로세스 보호", "french_switch": "Protection des Processus", "spanish_switch": "Protección de Procesos", "hindi_switch": "प्रक्रिया सुरक्षा", "arabic_switch": "حماية العمليات", "russian_switch": "Защита процессов", "slovenian_switch": "Zaščita procesov", }, "Process DLL Block": { "traditional_switch": "記憶體防護", "simplified_switch": "内存防护", "english_switch": "Memory Protection", "japanese_switch": "メモリ保護", "korean_switch": "메모리 보호", "french_switch": "Protection de la mémoire", "spanish_switch": "Protección de memoria", "hindi_switch": "मेमोरी सुरक्षा", "arabic_switch": "حماية الذاكرة", "russian_switch": "Защита памяти", "slovenian_switch": "Zaščita pomnilnika", }, "File Block": { "traditional_switch": "檔案防護", "simplified_switch": "文件防护", "english_switch": "File Protection", "japanese_switch": "ファイル保護", "korean_switch": "파일 보호", "french_switch": "Protection des Fichiers", "spanish_switch": "Protección de Archivos", "hindi_switch": "फ़ाइल सुरक्षा", "arabic_switch": "حماية الملفات", "russian_switch": "Защита файлов", "slovenian_switch": "Zaščita datotek", }, "Network Block": { "traditional_switch": "網路防護", "simplified_switch": "网络防护", "english_switch": "Network Protection", "japanese_switch": "ネットワーク保護", "korean_switch": "네트워크 보호", "french_switch": "Protection Réseau", "spanish_switch": "Protección de Red", "hindi_switch": "नेटवर्क सुरक्षा", "arabic_switch": "حماية الشبكة", "russian_switch": "Сетевая защита", "slovenian_switch": "Omrežna zaščita", }, "Driver Block": { "traditional_switch": "驅動防護", "simplified_switch": "驱动防护", "english_switch": "Driver Protection", "japanese_switch": "ドライバー保護", "korean_switch": "드라이버 보호", "french_switch": "Protection des Pilotes", "spanish_switch": "Protección de Controladores", "hindi_switch": "ड्राइवर सुरक्षा", "arabic_switch": "حماية برامج التشغيل", "russian_switch": "Защита драйверов", "slovenian_switch": "Zaščita gonilnikov", }, } path = source or "" messages = { "traditional_switch": f"威脅已終止: {path}", "simplified_switch": f"威胁已终止: {path}", "english_switch": f"Threat terminated: {path}", "japanese_switch": f"脅威が終了しました: {path}", "korean_switch": f"위협이 종료되었습니다: {path}", "french_switch": f"Menace terminée : {path}", "spanish_switch": f"Amenaza terminada: {path}", "hindi_switch": f"खतरा समाप्त: {path}", "arabic_switch": f"تم إنهاء التهديد: {path}", "russian_switch": f"Угроза устранена: {path}", "slovenian_switch": f"Grožnja odpravljena: {path}", } title = self._loc(titles.get(action, {})) message = self._loc(messages) try: self.tray_icon.notify(message, title) except Exception: log_exception("PYAS._MainMixin.trigger_block_notification:483") pass def show_notification(self, title, message): try: if self.tray_icon: self.tray_icon.notify(message, title) except Exception as e: log_exception("PYAS._MainMixin.show_notification:491") self.write_log("WARN", "show_notification", detail=str(e), success=False) def show_alert(self, title, message, style="info"): flags = 0x00000000 | ( 0x00000010 if style == "error" else 0x00000030 if style == "warning" else 0x00000040 ) self.user32.MessageBoxW(0, message, title, flags) return True def show_confirm(self, title, message): return self.user32.MessageBoxW(0, message, title, 0x00000004 | 0x00000020) == 6 def register_context_menu(self, enable): paths = [ r"Software\Classes\*\shell\PYAS_Scan", r"Software\Classes\Directory\shell\PYAS_Scan", ] cmd_path = ( f'"{self.file_pyas}"' if getattr(sys, "frozen", False) else f'"{self.python}" "{self.file_pyas}"' ) try: success = True for path in paths: if enable: success = ( self._reg_write( winreg.HKEY_CURRENT_USER, path, None, winreg.REG_SZ, "PYAS Security Scan", ) and success ) success = ( self._reg_write( winreg.HKEY_CURRENT_USER, path, "Icon", winreg.REG_SZ, f"{cmd_path},0" ) and success ) success = ( self._reg_write( winreg.HKEY_CURRENT_USER, rf"{path}\command", None, winreg.REG_SZ, f'{cmd_path} -scan "%1"', ) and success ) else: for entry in (rf"{path}\command", path): try: with winreg.OpenKey( winreg.HKEY_CURRENT_USER, entry, 0, winreg.KEY_READ ): pass except FileNotFoundError: continue success = self._reg_delete(winreg.HKEY_CURRENT_USER, entry) and success if not success: self.write_log( "WARN", "register_context_menu", detail="Registry operation failed", success=False, ) return success except Exception as error: log_exception("PYAS._MainMixin.register_context_menu:523") self.write_log("WARN", "register_context_menu", detail=str(error), success=False) return False def trigger_context_scan(self, target): if self._window: self._window.evaluate_js( f"if(window.triggerContextScan) window.triggerContextScan({json.dumps(target.replace(os.sep, '/'))});" ) def on_drop(self, e): def _process_drop(): try: files = e.get("dataTransfer", {}).get("files", []) paths = [f.get("pywebviewFullPath") for f in files if f.get("pywebviewFullPath")] if paths and self._window: self._window.evaluate_js( f"if(window.triggerContextScan) window.triggerContextScan({json.dumps(paths)});" ) except Exception as ex: log_exception("PYAS._MainMixin.on_drop._process_drop:540") self.write_log("WARN", "on_drop", detail=str(ex), success=False) threading.Thread(target=_process_drop, daemon=True).start() def select_files(self, file_types=None): if self._window: kwargs = {"allow_multiple": True} if file_types: kwargs["file_types"] = tuple(file_types) return ( self._window.create_file_dialog(getattr(webview, "OPEN_DIALOG", 10), **kwargs) or [] ) return [] def select_folder(self): if self._window: return self._window.create_file_dialog(getattr(webview, "FOLDER_DIALOG", 20)) or [] return [] def open_file_location(self, file_path): if not file_path: return False expanded_path = os.path.expandvars(file_path).strip('"').strip("'") reg_prefixes = ("HKLM", "HKCU", "HKCR", "HKU", "HKCC", "HKEY_") if expanded_path.upper().startswith(reg_prefixes): try: full_path = expanded_path if full_path.startswith("HKLM"): full_path = full_path.replace("HKLM", "HKEY_LOCAL_MACHINE", 1) elif full_path.startswith("HKCU"): full_path = full_path.replace("HKCU", "HKEY_CURRENT_USER", 1) elif full_path.startswith("HKCR"): full_path = full_path.replace("HKCR", "HKEY_CLASSES_ROOT", 1) elif full_path.startswith("HKU"): full_path = full_path.replace("HKU", "HKEY_USERS", 1) elif full_path.startswith("HKCC"): full_path = full_path.replace("HKCC", "HKEY_CURRENT_CONFIG", 1) self._run_windows_tool( ("taskkill.exe",), ["/F", "/IM", "regedit.exe"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) self._reg_write( winreg.HKEY_CURRENT_USER, r"Software\Microsoft\Windows\CurrentVersion\Applets\Regedit", "LastKey", winreg.REG_SZ, full_path, ) regedit = self._find_windows_tool("regedit.exe") if not regedit: return False subprocess.Popen([regedit]) return True except Exception: log_exception("PYAS._MainMixin.open_file_location:597") pass return False if os.path.exists(expanded_path): try: clean_path = os.path.normpath(expanded_path) explorer = self._find_windows_tool("explorer.exe") if not explorer: return False subprocess.Popen([explorer, "/select,", clean_path]) return True except Exception: log_exception("PYAS._MainMixin.open_file_location:610") pass return False def open_website(self): try: return webbrowser.open(self.pyas_config.get("api_host")) except Exception: log_exception("PYAS._MainMixin.open_website:617") return False def open_url(self, url): try: return webbrowser.open(url) except Exception: log_exception("PYAS._MainMixin.open_url:623") return False class WindowAPI(_MainMixin, ScannerMixin, ToolsMixin, ProtectMixin): pass def get_base_path(): if getattr(sys, "frozen", False): return os.path.dirname(sys.executable) return os.path.dirname(os.path.abspath(__file__)) def get_frontend_asset_errors(): base_path = get_base_path() required_files = [ os.path.join("Interface", "templates", "index.html"), os.path.join("Interface", "static", "css", "style.css"), os.path.join("Interface", "static", "js", "i18n.js"), os.path.join("Interface", "static", "js", "main.js"), ] errors = [] for rel_path in required_files: abs_path = os.path.join(base_path, rel_path) if not os.path.isfile(abs_path): errors.append(f"Missing UI asset: {rel_path}") elif os.path.getsize(abs_path) <= 0: errors.append(f"Empty UI asset: {rel_path}") return errors def show_startup_error(message): try: ctypes.windll.user32.MessageBoxW(None, message, PYAS_WINDOW_TITLE, 0x00000010) except Exception: log_exception("PYAS.show_startup_error:662") pass class NoCacheRequestHandler(SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=os.path.join(get_base_path(), "Interface"), **kwargs) def log_message(self, format, *args): logging.getLogger("PYAS.Startup").info("HTTP %s: " + format, self.client_address[0], *args) def do_GET(self): if self.path == "/": self.path = "/templates/index.html" return super().do_GET() def end_headers(self): self.send_header("Cache-Control", "no-cache, no-store, must-revalidate") self.send_header("Expires", "0") self.send_header("Pragma", "no-cache") super().end_headers() class WindowHook: def __init__(self, title, api_ref=None): self.title, self.api_ref = title, api_ref self.old_wndproc = None self.WM_DPICHANGED = 0x02E0 self.WM_NCHITTEST = 0x0084 self.WM_CLOSE = 0x0010 self.WM_COPYDATA = WM_COPYDATA self.WM_SYSCOMMAND = 0x0112 self.SC_CLOSE = 0xF060 self.HTCAPTION = 2 self.GWLP_WNDPROC = -4 self.RECT = RECT self.WNDPROC = ctypes.WINFUNCTYPE( ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p ) self.user32 = ctypes.windll.user32 self.user32.FindWindowW.argtypes, self.user32.FindWindowW.restype = [ ctypes.c_wchar_p, ctypes.c_wchar_p, ], ctypes.wintypes.HWND self.user32.CallWindowProcW.argtypes, self.user32.CallWindowProcW.restype = [ ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p, ], ctypes.c_void_p self.user32.DefWindowProcW.argtypes, self.user32.DefWindowProcW.restype = [ ctypes.c_void_p, ctypes.c_uint, ctypes.c_void_p, ctypes.c_void_p, ], ctypes.c_void_p if ctypes.sizeof(ctypes.c_void_p) == 8: self.SetWindowLong, self.GetWindowLong = ( self.user32.SetWindowLongPtrW, self.user32.GetWindowLongPtrW, ) else: self.SetWindowLong, self.GetWindowLong = ( self.user32.SetWindowLongW, self.user32.GetWindowLongW, ) self.SetWindowLong.argtypes, self.SetWindowLong.restype = [ ctypes.c_void_p, ctypes.c_int, self.WNDPROC, ], ctypes.c_void_p self.GetWindowLong.argtypes, self.GetWindowLong.restype = [ ctypes.c_void_p, ctypes.c_int, ], ctypes.c_void_p self.new_wndproc_cb = self.WNDPROC(self.wndproc) def hook(self): if self.old_wndproc: return hwnd = self.user32.FindWindowW(None, self.title) if hwnd: self.old_wndproc = self.GetWindowLong(hwnd, self.GWLP_WNDPROC) self.SetWindowLong(hwnd, self.GWLP_WNDPROC, self.new_wndproc_cb) self.maintenance_channel_ready = False try: self.user32.ChangeWindowMessageFilterEx.argtypes = [ ctypes.c_void_p, ctypes.c_uint, ctypes.c_uint, ctypes.c_void_p, ] self.user32.ChangeWindowMessageFilter.argtypes = [ctypes.c_uint, ctypes.c_uint] self.user32.ChangeWindowMessageFilter.restype = ctypes.wintypes.BOOL self.user32.ChangeWindowMessageFilter(PYAS_MAINTENANCE_MESSAGE, MSGFLT_REMOVE) filter_info = (ctypes.wintypes.DWORD * 2)(8, 0) if not self.user32.ChangeWindowMessageFilterEx( hwnd, PYAS_MAINTENANCE_MESSAGE, MSGFLT_DISALLOW, ctypes.byref(filter_info) ): raise OSError("Could not protect maintenance message channel") if filter_info[1] == 3: raise OSError("Maintenance message remains allowed by a wider filter") self.maintenance_channel_ready = True self.user32.ChangeWindowMessageFilterEx(hwnd, self.WM_COPYDATA, 1, None) except Exception: log_exception("PYAS.WindowHook.hook:732") pass def call_default(self, hwnd, msg, wparam, lparam): if self.old_wndproc: return self.user32.CallWindowProcW(self.old_wndproc, hwnd, msg, wparam, lparam) return self.user32.DefWindowProcW(hwnd, msg, wparam, lparam) def wndproc(self, hwnd, msg, wparam, lparam): if msg == PYAS_MAINTENANCE_MESSAGE: if ( not getattr(self, "maintenance_channel_ready", False) or wparam not in PYAS_MAINTENANCE_COMMANDS or not self.api_ref ): return 0 uninstall = wparam in (PYAS_MESSAGE_QUIT, PYAS_MESSAGE_DRIVER_UNINSTALL) threading.Thread( target=self.api_ref.close, kwargs={"uninstall_driver": uninstall}, daemon=True ).start() return 1 if msg == self.WM_COPYDATA: try: if not lparam: return 0 cds = COPYDATASTRUCT.from_address(lparam) if cds.dwData not in (PYAS_MESSAGE_SCAN, PYAS_MESSAGE_SHOW): return 0 if cds.dwData == PYAS_MESSAGE_SCAN and ( not cds.lpData or not 0 < cds.cbData <= 131072 ): return 0 if cds.dwData == PYAS_MESSAGE_SCAN: path = ctypes.string_at(cds.lpData, cds.cbData).decode("utf-8").strip("\x00") if self.api_ref: threading.Thread(target=self.api_ref.restore_from_tray, daemon=True).start() threading.Thread( target=self.api_ref.trigger_context_scan, args=(path,), daemon=True ).start() elif cds.dwData == PYAS_MESSAGE_SHOW and self.api_ref: threading.Thread(target=self.api_ref.restore_from_tray, daemon=True).start() except Exception: log_exception("PYAS.WindowHook.wndproc:762") pass return 1 if msg == self.WM_CLOSE or ( msg == self.WM_SYSCOMMAND and (wparam & 0xFFF0) == self.SC_CLOSE ): if self.api_ref and not getattr(self.api_ref, "closing", False): threading.Thread(target=self.api_ref.hide_window, daemon=True).start() return 0 return self.call_default(hwnd, msg, wparam, lparam) if msg == self.WM_NCHITTEST: x, y = lparam & 0xFFFF, (lparam >> 16) & 0xFFFF if x >= 32768: x -= 65536 if y >= 32768: y -= 65536 rect = self.RECT() self.user32.GetWindowRect(hwnd, ctypes.byref(rect)) if rect.top <= y <= rect.top + 44 and rect.left <= x <= rect.right - 150: return self.HTCAPTION if msg == self.WM_DPICHANGED: try: rect = self.RECT.from_address(lparam) self.user32.SetWindowPos( hwnd, None, rect.left, rect.top, rect.right - rect.left, rect.bottom - rect.top, 0x0004 | 0x0010 | 0x0020, ) except Exception: log_exception("PYAS.WindowHook.wndproc:788") pass return self.call_default(hwnd, msg, wparam, lparam) class StartupHTTPServer(ThreadingTCPServer): allow_reuse_address = True daemon_threads = True def handle_error(self, request, client_address): logging.getLogger("PYAS.Startup").exception("HTTP request failed: %s", client_address) def start_api(port_container, error_container, ready_event): logger = logging.getLogger("PYAS.Startup") try: with StartupHTTPServer(("127.0.0.1", 0), NoCacheRequestHandler) as httpd: port_container.append(httpd.server_address[1]) logger.info("HTTP listener ready: 127.0.0.1:%s", port_container[0]) ready_event.set() httpd.serve_forever() except Exception as e: logger.exception("HTTP listener failed") error_container.append(str(e)) ready_event.set() def verify_ui_server(port): from urllib.request import build_opener, ProxyHandler opener = build_opener(ProxyHandler({})) for path in ("/", "/static/css/style.css", "/static/js/i18n.js", "/static/js/main.js"): with opener.open(f"http://127.0.0.1:{port}{path}", timeout=3) as response: if response.status != 200 or not response.read(1): raise RuntimeError(f"UI HTTP health check failed: {path}") logging.getLogger("PYAS.Startup").info("HTTP UI health checks passed") def start_ui(): logger = logging.getLogger("PYAS.Startup") hide_on_start = "-h" in sys.argv or "-hide" in sys.argv init_width, init_height = 980, 670 stage = "assets" js_api = None webview_completed = threading.Event() recovery_lock = threading.Lock() def fail_startup(reason, recover=False): with recovery_lock: if webview_completed.is_set(): return webview_completed.set() logger.error("Startup failed: %s", reason) if js_api: js_api.write_log("WARN", "WebView2", detail=reason, success=False) js_api.flush_logs_now() if ( recover and js_api and restart_with_new_webview_profile(js_api.path_webview, __file__) ): os._exit(3) if not hide_on_start: show_startup_error( f"PYAS could not initialize its interface.\n\n{reason}\n\n" f"Diagnostic log: {STARTUP_LOG_PATH or 'unavailable'}" ) os._exit(3) try: wait_for_recovery_parent() arguments = os.environ.get("WEBVIEW2_ADDITIONAL_BROWSER_ARGUMENTS", "") os.environ["WEBVIEW2_ADDITIONAL_BROWSER_ARGUMENTS"] = ( arguments + " --proxy-bypass-list=localhost;127.0.0.1" ).strip() frontend_errors = get_frontend_asset_errors() if frontend_errors: raise RuntimeError("UI files are incomplete: " + "; ".join(frontend_errors)) logger.info("UI assets verified") stage = "http" port_container, server_errors = [], [] server_ready = threading.Event() threading.Thread( target=start_api, args=(port_container, server_errors, server_ready), daemon=True ).start() if not server_ready.wait(5.0) or not port_container or server_errors: raise RuntimeError("Local UI server did not start: " + "; ".join(server_errors)) verify_ui_server(port_container[0]) stage = "api" logger.info("Initializing Python API") js_api = WindowAPI() js_api.file_webview_log = STARTUP_LOG_PATH or js_api.file_webview_log stage = "profile" preferred_profile = os.environ.get("PYAS_WEBVIEW_PROFILE", js_api.path_webview) js_api.path_webview = prepare_webview_profile(preferred_profile) stage = "runtime" check_webview_dependencies(webview) stage = "window" user32 = ctypes.windll.user32 pos_x = (user32.GetSystemMetrics(0) - init_width) // 2 pos_y = (user32.GetSystemMetrics(1) - init_height) // 2 startup_url = f"http://127.0.0.1:{port_container[0]}/" window = webview.create_window( title=PYAS_WINDOW_TITLE, url=startup_url, width=init_width, height=init_height, x=pos_x, y=pos_y, frameless=True, easy_drag=False, js_api=js_api, background_color="#e0e0e0", hidden=hide_on_start, ) if platform.system() == "Windows": window_hook = WindowHook(PYAS_WINDOW_TITLE, js_api) window.events.shown += window_hook.hook js_api.set_window(window) logger.info("Native window configured; hidden=%s", hide_on_start) stage = "tray" js_api.show_tray() logger.info("Tray initialized") webview_loaded_event = threading.Event() dnd_state = {"bound": False} def bind_dnd(): webview_loaded_event.set() logger.info("WebView document loaded") if dnd_state["bound"]: return try: window.dom.document.events.drop += DOMEventHandler(js_api.on_drop, True, True) dnd_state["bound"] = True except Exception: logger.exception("Drag and drop binding failed") def webview_startup_watchdog(): if js_api.ui_ready_event.wait(30.0) or webview_completed.is_set(): return if webview_loaded_event.is_set(): logger.warning("UI bridge timeout; reloading once") try: window.load_url(startup_url) except Exception: logger.exception("UI reload failed") if js_api.ui_ready_event.wait(30.0) or webview_completed.is_set(): return reason = ( "Python bridge or configuration initialization timed out" if webview_loaded_event.is_set() else ("WebView2 did not load the local UI document") ) fail_startup(reason, recover=True) def on_closed(): webview_completed.set() window.events.loaded += bind_dnd window.events.closed += on_closed threading.Thread(target=webview_startup_watchdog, daemon=True).start() stage = "webview" logger.info("Starting edgechromium") webview.start(gui="edgechromium", private_mode=False, storage_path=js_api.path_webview) webview_completed.set() except Exception as e: logger.exception("Startup exception at stage=%s", stage) fail_startup(f"{stage}: {e}", recover=stage == "webview") if __name__ == "__main__": if "-quit" in sys.argv or "-driver-unload" in sys.argv or "-driver-uninstall" in sys.argv: controller = WindowAPI() if "-quit" in sys.argv or "-driver-uninstall" in sys.argv: success, delete_error = controller.uninstall_system_driver() if not success: controller.write_log( "WARN", "Driver Service", detail=f"DeleteService failed: 0x{delete_error & 0xFFFFFFFF:08X}", success=False, ) else: success = controller.stop_system_driver() controller.flush_logs_now() os._exit(0 if success else 2) start_ui()