diff --git a/PYAS.py b/PYAS.py new file mode 100644 index 0000000..666055c --- /dev/null +++ b/PYAS.py @@ -0,0 +1,1561 @@ +import os, gc, re, sys, time, json, base64 +import ctypes, ctypes.wintypes, threading +import requests, pyperclip, winreg, shutil +import traceback, hashlib, msvcrt + +from PYAS_Config import * +from PYAS_Engine import * +from PYAS_Interface import * + +from PySide6.QtWidgets import * +from PySide6.QtCore import * +from PySide6.QtGui import * + +#################################################################################################### + +class PROCESSENTRY32(ctypes.Structure): + _fields_ = [ + ("dwSize", ctypes.wintypes.DWORD), + ("cntUsage", ctypes.wintypes.DWORD), + ("th32ProcessID", ctypes.wintypes.DWORD), + ("th32DefaultHeapID", ctypes.wintypes.LPVOID), + ("th32ModuleID", ctypes.wintypes.DWORD), + ("cntThreads", ctypes.wintypes.DWORD), + ("th32ParentProcessID", ctypes.wintypes.DWORD), + ("dwFlags", ctypes.wintypes.DWORD), + ("szExeFile", ctypes.wintypes.CHAR * 260)] + +class MIB_TCPROW_OWNER_PID(ctypes.Structure): + _fields_ = [ + ("dwState", ctypes.wintypes.DWORD), + ("dwLocalAddr", ctypes.wintypes.DWORD), + ("dwLocalPort", ctypes.wintypes.DWORD), + ("dwRemoteAddr", ctypes.wintypes.DWORD), + ("dwRemotePort", ctypes.wintypes.DWORD), + ("dwOwningPid", ctypes.wintypes.DWORD)] + +class MIB_TCPTABLE_OWNER_PID(ctypes.Structure): + _fields_ = [ + ("dwNumEntries", ctypes.wintypes.DWORD), + ("table", MIB_TCPROW_OWNER_PID * 1)] + +class FILE_NOTIFY_INFORMATION(ctypes.Structure): + _fields_ = [ + ("NextEntryOffset", ctypes.wintypes.DWORD), + ("Action", ctypes.wintypes.DWORD), + ("FileNameLength", ctypes.wintypes.DWORD), + ("FileName", ctypes.wintypes.WCHAR * 1024)] + +class SERVICE_STATUS(ctypes.Structure): + _fields_ = [ + ("dwServiceType", ctypes.wintypes.DWORD), + ("dwCurrentState", ctypes.wintypes.DWORD), + ("dwControlsAccepted", ctypes.wintypes.DWORD), + ("dwWin32ExitCode", ctypes.wintypes.DWORD), + ("dwServiceSpecificExitCode", ctypes.wintypes.DWORD), + ("dwCheckPoint", ctypes.wintypes.DWORD), + ("dwWaitHint", ctypes.wintypes.DWORD),] + +#################################################################################################### + +class MainWindow_Controller(QMainWindow): + scan_progress_signal = Signal(str) + scan_add_virus_signal = Signal(str) + scan_result_signal = Signal(str) + scan_reset_signal = Signal() + progress_title_signal = Signal(str) + + def __init__(self): + super().__init__() + self.setAttribute(Qt.WA_TranslucentBackground) + self.setWindowFlags(Qt.FramelessWindowHint) + self.init_environ() + self.init_variable() + self.load_config() + self.init_interface() + self.init_windll() + self.backup_mbr() + self.relock_file() + self.init_whitelist() + self.init_thread() + self.init_connect() + self.show_startup() + +#################################################################################################### + + def init_variable(self): + self.ui = Ui_main_window() + self.ui.setupUi(self) + self.python = sys.executable + self.process_timer = QTimer(self) + + self.model = model_scanner() + self.model.load_path(self.path_models) + self.rule = rule_scanner() + self.rule.load_path(self.path_rules) + + self.mouse_flag = 0 + self.mouse_pos = 0 + self.current_opacity = 0 + + self.scan_running = False + self.virus_lock = {} + self.virus_results = [] + self.scan_count = 0 + self.mbr_backup = {} + + self.theme_names = [ + "white_switch", "red_switch", "yellow_switch", + "green_switch", "blue_switch", "black_switch" + ] + self.lang_names = [ + "traditional_switch", "simplified_switch", "english_switch" + ] + self.pyas_default = { + "version": "3.3.1", + "product": "00000-00000-00000-00000-00000", + "language": "english_switch", + "theme": "white_switch", + "sensitive": 95, + "custom_rule": {}, + "process_switch": True, + "document_switch": True, + "system_switch": True, + "driver_switch": True, + "network_switch": True, + } + self.pass_windows = [ + {"exe": "System Idle Process", "class": "", "title": ""}, + {"exe": "", "class": "Windows.UI.Core.CoreWindow", "title": ""}, + {"exe": "", "class": "Qt691QWindowIcon", "title": "PYAS"}, + {"exe": "explorer.exe", "class": "", "title": ""}, + ] + self.block_replace = { + "MBR_BLOCK": "引導行為攔截", + "REG_BLOCK": "註冊表行為攔截", + "FILE_BLOCK": "檔案行為攔截" + } + +#################################################################################################### + + def init_environ(self): + self.file_pyas = self.norm_path(sys.argv[0]) + self.args_pyas = sys.argv[1:] + self.path_pyas = os.path.dirname(self.file_pyas) + + self.path_appdata = os.environ.get("APPDATA") + self.path_name = os.environ.get("USERNAME") + self.path_temp = os.environ.get("TEMP", f"C:\\Users\\{self.path_name}\\AppData\\Local\\Temp") + self.path_config = os.environ.get("ALLUSERSPROFILE", "C:\\ProgramData") + self.path_system = os.environ.get("SYSTEMROOT", "C:\\Windows") + self.path_user = os.environ.get("USERPROFILE", f"C:\\Users\\{self.path_name}") + + self.path_systemp = os.path.join(self.path_system, "Temp") + self.file_config = os.path.join(self.path_config, "PYAS", "Config.json") + self.path_models = os.path.join(self.path_pyas, "Engine", "Models") + self.path_rules = os.path.join(self.path_pyas, "Engine", "Rules") + self.path_protect = os.path.join(self.path_pyas, "Plugins", "Filter") + self.path_drivers = os.path.join(self.path_protect, "PYAS_Driver.sys") + +#################################################################################################### + + def init_windll(self): + for name in ["ntdll", "Psapi", "user32", "kernel32", "advapi32", "iphlpapi"]: + try: + setattr(self, name.lower(), ctypes.WinDLL(name, use_last_error=True)) + except Exception as e: + self.send_message(e, "warn", False) + + self.advapi32.OpenSCManagerW.argtypes = [ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.wintypes.DWORD] + self.advapi32.OpenSCManagerW.restype = ctypes.wintypes.HANDLE + + self.advapi32.CreateServiceW.argtypes = [ + ctypes.wintypes.HANDLE, ctypes.c_wchar_p, ctypes.c_wchar_p, ctypes.wintypes.DWORD, + ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.wintypes.DWORD, ctypes.c_wchar_p, + ctypes.c_wchar_p, ctypes.POINTER(ctypes.wintypes.DWORD), ctypes.c_wchar_p, ctypes.c_wchar_p, + ctypes.c_wchar_p + ] + self.advapi32.CreateServiceW.restype = ctypes.wintypes.HANDLE + + self.advapi32.OpenServiceW.argtypes = [ctypes.wintypes.HANDLE, ctypes.c_wchar_p, ctypes.wintypes.DWORD] + self.advapi32.OpenServiceW.restype = ctypes.wintypes.HANDLE + + self.advapi32.StartServiceW.argtypes = [ctypes.wintypes.HANDLE, ctypes.wintypes.DWORD, ctypes.POINTER(ctypes.c_wchar_p)] + self.advapi32.StartServiceW.restype = ctypes.wintypes.BOOL + + self.advapi32.ControlService.argtypes = [ctypes.wintypes.HANDLE, ctypes.wintypes.DWORD, ctypes.POINTER(SERVICE_STATUS)] + self.advapi32.ControlService.restype = ctypes.wintypes.BOOL + + self.advapi32.DeleteService.argtypes = [ctypes.wintypes.HANDLE] + self.advapi32.DeleteService.restype = ctypes.wintypes.BOOL + + self.advapi32.CloseServiceHandle.argtypes = [ctypes.wintypes.HANDLE] + self.advapi32.CloseServiceHandle.restype = ctypes.wintypes.BOOL + +#################################################################################################### + + def save_config(self, file_path, config): + try: + filter_config = {} + for k, v in config.items(): + if k.endswith("_button") and isinstance(v, bool) and k.replace("_button", "_window") in self.widgets: + continue + filter_config[k] = v + path = os.path.dirname(file_path) + if not os.path.exists(path): + os.makedirs(path) + with open(file_path, "w", encoding="utf-8", errors="ignore") as f: + f.write(json.dumps(filter_config, indent=4, ensure_ascii=False)) + except Exception as e: + self.send_message(e, "warn", False) + + def read_config(self, file_path, default_value): + try: + config_value = {} + if os.path.exists(file_path): + with open(file_path, "r", encoding="utf-8", errors="ignore") as f: + config_value = json.load(f) + for key, value in default_value.items(): + config_value.setdefault(key, value) + return config_value + except Exception as e: + self.send_message(e, "warn", False) + return default_value.copy() + + def load_config(self): + self.pyas_config = self.read_config(self.file_config, self.pyas_default) + + def config_list(self, key): + if key not in self.pyas_config: + self.pyas_config[key] = [] + return self.pyas_config[key] + +#################################################################################################### + + def init_interface(self): + self.widgets = {k: v for k, v in self.ui.__dict__.items() if isinstance(v, QWidget)} + + self.menu_window = [self.widgets.get(x) for x in ["options_window", "navigate_window"]] + self.sub_window = [self.widgets.get(x) for x in ["method_window", "solve_button"]] + self.blank_window = [self.widgets.get("blank_window")] + self.main_window = [self.widgets.get(x) for x in [ + "about_window", "setting_window", "protect_window", + "taskmgr_window", "tool_window", "scan_window", "home_window"]] + + for w in self.sub_window: + if w: w.hide() + for w in self.main_window + self.menu_window: + if w: w.raise_() + for w in self.menu_window + self.blank_window: + if w: w.setGraphicsEffect(self.create_shadow()) + + self.last_widget = self.widgets.get("home_window") + self.setup_theme_mapping() + self.setup_process_menu() + + def create_shadow(self): + shadow = QGraphicsDropShadowEffect(self) + shadow.setOffset(0, 0) + shadow.setBlurRadius(20) + shadow.setColor(QColor(175, 175, 175)) + return shadow + +#################################################################################################### + + def init_connect(self): + self.tray_icon = QSystemTrayIcon(self) + self.tray_icon.setIcon(QIcon(r":/icon/logo_black.png")) + self.tray_icon.activated.connect(self.show_button) + self.tray_icon.show() + + self.scan_progress_signal.connect(self.slot_scan_progress) + self.scan_add_virus_signal.connect(self.slot_scan_add_virus) + self.scan_result_signal.connect(self.slot_scan_result) + self.scan_reset_signal.connect(self.slot_scan_reset) + self.progress_title_signal.connect(self.slot_progress_title) + + for name, widget in self.widgets.items(): + if hasattr(widget, "setCheckable") and widget.isCheckable(): + if name in self.lang_names: + widget.toggled.connect(lambda checked, ln=name: self.on_language_radio(ln, checked)) + if self.pyas_config.get("language") == name: + widget.setChecked(True) + elif name in self.theme_names: + widget.toggled.connect(lambda checked, tn=name: self.on_theme_radio(tn, checked)) + if self.pyas_config.get("theme") == name: + widget.setChecked(True) + elif "_switch" in name: + widget.toggled.connect(lambda checked, n=name: self.save_state(n, checked)) + if name in self.pyas_config: + try: + widget.setChecked(self.pyas_config[name]) + except Exception: + pass + + if "_button" in name and hasattr(widget, "clicked"): + wn = name.replace("_button", "_window") + if self.widgets.get(wn): + widget.clicked.connect(lambda _, wn=wn: self.change_window(wn)) + elif hasattr(self, name): + widget.clicked.connect(getattr(self, name)) + + def on_language_radio(self, ln, checked): + if checked: + self.pyas_config["language"] = ln + self.save_config(self.file_config, self.pyas_config) + self.apply_settings() + + def on_theme_radio(self, tn, checked): + if checked: + self.pyas_config["theme"] = tn + self.save_config(self.file_config, self.pyas_config) + self.apply_settings() + +#################################################################################################### + + def save_state(self, name, checked): + self.pyas_config[name] = bool(checked) + if name == "sensitive_switch": + self.pyas_config["sensitive"] = 0 if checked else 95 + elif name == "extension_switch": + pass + elif name == "process_switch": + if checked: + self.start_daemon_thread(self.protect_proc_thread) + elif name == "document_switch": + if checked: + self.start_daemon_thread(self.protect_file_thread) + elif name == "system_switch": + if checked: + self.start_daemon_thread(self.protect_system_thread) + elif name == "driver_switch": + if checked: + self.install_system_driver() + self.start_daemon_thread(self.pipe_server_thread) + else: + self.stop_system_driver() + elif name == "network_switch": + if checked: + self.start_daemon_thread(self.protect_net_thread) + else: + self.send_message(f"此功能不支持使用", "info", True) + self.save_config(self.file_config, self.pyas_config) + + def init_thread(self): + self.start_daemon_thread(self.popup_intercept_thread) + + def start_daemon_thread(self, target, *args, **kwargs): + t = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=True) + t.start() + return t + +#################################################################################################### + + def setup_theme_mapping(self): + for name, widget in self.widgets.items(): + mapping = {} + base_name = name.replace("_button", "").replace("_window", "") + theme_icon_map = { + theme: f":/icon/{base_name}_{'white' if theme == 'black_switch' else 'black'}.png" + for theme in self.theme_names} + + if hasattr(widget, "setPixmap"): + mapping["pixmap"] = theme_icon_map + if hasattr(widget, "setIcon"): + mapping["icon"] = theme_icon_map + if mapping: + widget._theme_mapping = mapping + +#################################################################################################### + + def apply_settings(self): + lang = self.pyas_config.get("language", "english_switch") + theme = self.pyas_config.get("theme", "white_switch") + lang_map = translate_dict.get(lang, {}) + theme_map = translate_dict.get(theme, {}) + + for name, widget in self.widgets.items(): + if "_switch" in name and hasattr(widget, "setText") and hasattr(widget, "isChecked"): + if name not in self.lang_names + self.theme_names: + checked = widget.isChecked() + widget.setText(lang_map.get("開啟", "開啟") if checked else lang_map.get("關閉", "關閉")) + + if not widget.property("_connected_toggle"): + widget.toggled.connect(lambda checked, b=widget: self.toggle_switch_text(b, checked)) + widget.setProperty("_connected_toggle", True) + + elif hasattr(widget, "setText"): + if not hasattr(widget, "_origin_text"): + widget._origin_text = self.get_widget_text(widget) + widget.setText(self.trans(lang, widget._origin_text)) + + if hasattr(widget, "setStyleSheet"): + if not hasattr(widget, "_origin_style"): + widget._origin_style = widget.styleSheet() + style = widget._origin_style + for k, v in theme_map.items(): + style = style.replace(str(k), str(v)) + widget.setStyleSheet(style) + + mapping = getattr(widget, "_theme_mapping", None) + if isinstance(mapping, dict): + if "icon" in mapping and hasattr(widget, "setIcon"): + icon_path = mapping["icon"].get(theme) + if icon_path and (os.path.exists(icon_path) or icon_path.startswith(":")): + widget.setIcon(QIcon(icon_path)) + widget.icon_path = icon_path + + if "pixmap" in mapping and hasattr(widget, "setPixmap"): + pm_path = mapping["pixmap"].get(theme) + if pm_path and pm_path.startswith(":"): + pix = QPixmap(pm_path) + if not pix.isNull(): + widget.setPixmap(pix) + widget.pixmap_path = pm_path + +#################################################################################################### + + def toggle_switch_text(self, btn, checked): + lang = self.pyas_config.get("language", "traditional_switch") + switch_map = translate_dict.get(lang, {}) + btn.setText(switch_map.get("開啟", "開啟") if checked else switch_map.get("關閉", "關閉")) + + def get_widget_text(self, widget): + for method in ["text", "toPlainText", "currentText"]: + if hasattr(widget, method) and callable(getattr(widget, method)): + return getattr(widget, method)() + return "" + + def trans(self, language, text): + for key, value in translate_dict.get(language, translate_dict).items(): + text = str(text).replace(str(key), str(value)) + return text.rstrip() + +#################################################################################################### + + def change_opacity(self, widget, current, target, duration=200): + self.cleanup_animation(widget, "opacity_animation") + anim = QPropertyAnimation(widget, b"windowOpacity") + anim.setStartValue(current / 100.0) + anim.setEndValue(target / 100.0) + anim.setDuration(duration) + anim.setEasingCurve(QEasingCurve.Linear) + anim.valueChanged.connect(lambda v: self.update_opacity(widget, v)) + anim.start() + widget.opacity_animation = anim + + def update_opacity(self, widget, value): + widget.setWindowOpacity(value) + widget.current_opacity = value * 100 + widget.setVisible(value > 0) + + def cleanup_animation(self, widget, attr): + anim = getattr(widget, attr, None) + if anim: + anim.stop() + anim.deleteLater() + setattr(widget, attr, None) + +#################################################################################################### + + def change_window(self, widget_name, duration=200): + widget = self.widgets.get(widget_name) + if widget and widget != self.last_widget: + widget.raise_() + widget.hide() + for window in self.menu_window: + window.raise_() + for window in self.sub_window: + window.hide() + self.animate_geometry(widget, duration) + self.last_widget = widget + + if widget_name == "taskmgr_window": + self.start_process_timer() + else: + self.stop_process_timer() + + def animate_geometry(self, widget, duration): + click = widget.mapFromGlobal(QCursor.pos()) + w, h = widget.width(), widget.height() + final_radius = int(((max(click.x(), w - click.x())) ** 2 + (max(click.y(), h - click.y())) ** 2) ** 0.5) + anim = QVariantAnimation(widget) + anim.setStartValue(0) + anim.setEndValue(final_radius) + anim.setDuration(duration) + anim.valueChanged.connect(lambda val: self.update_geometry(widget, click, val)) + anim.start() + + def update_geometry(self, widget, click, value): + widget.setMask(QRegion(click.x() - value, click.y() - value, 2 * value, 2 * value, QRegion.Ellipse)) + widget.setVisible(value > 0) + +#################################################################################################### + + def mousePressEvent(self, event: QMouseEvent): + if event.button() == Qt.LeftButton: + self.mouse_flag = True + self.mouse_pos = event.globalPosition().toPoint() - self.pos() + self.change_opacity(self, self.current_opacity, 80, 100) + event.accept() + + def mouseMoveEvent(self, event: QMouseEvent): + if self.mouse_flag: + new_top_left = event.globalPosition().toPoint() - self.mouse_pos + self.move(new_top_left) + event.accept() + + def mouseReleaseEvent(self, event: QMouseEvent): + if event.button() == Qt.LeftButton and self.mouse_flag: + self.mouse_flag = False + self.setCursor(QCursor(Qt.ArrowCursor)) + self.change_opacity(self, self.current_opacity, 100, 100) + event.accept() + +#################################################################################################### + + def nativeEvent(self, eventType, message): + event = ctypes.wintypes.MSG.from_address(int(message)) + if event.message in [0x0010, 0x0002, 0x0012, 0x0112, 0x0212]: + return True, 0 + return super().nativeEvent(eventType, message) + + def show_startup(self): + try: + if self.singleton_mutex("pyas_security"): + param = "" + if self.args_pyas: + param = self.args_pyas[0].replace("/", "-") + if "-h" not in param: + self.show_button() + self.apply_settings() + else: + sys.exit(0) + except Exception as e: + self.send_message(e, "warn", False) + + def show_button(self): + self.show() + self.change_opacity(self, self.current_opacity, 100, 200) + + def minimize_button(self): + self.change_opacity(self, self.current_opacity, 0, 200) + + def close_button(self): + if self.send_message("您確定要退出所有防護嗎?", "quest", True): + 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.stop_system_driver() + + self.minimize_button() + while self.isVisible(): + QApplication.processEvents() + if hasattr(self, "tray_icon"): + self.tray_icon.hide() + self.tray_icon.deleteLater() + QApplication.quit() + + def reset_button(self): + if self.send_message("您確定要重置所有設定嗎?", "quest", True): + self.pyas_config = self.pyas_default.copy() + self.restart_button() + + def restart_button(self): + self.save_config(self.file_config, self.pyas_config) + self.minimize_button() + while self.isVisible(): + QApplication.processEvents() + os.execl(self.python, self.python, *sys.argv) + + def singleton_mutex(self, name): + mutex = self.kernel32.CreateMutexW(None, False, name) + if self.kernel32.GetLastError() == 183: + return False + return True + +#################################################################################################### + + def send_message(self, message, mode, translate=True): + if QThread.currentThread() != QApplication.instance().thread(): + QMetaObject.invokeMethod(self, "send_message_thread", Qt.QueuedConnection, + Q_ARG("QString", str(message)), Q_ARG("QString", mode), Q_ARG("bool", translate)) + return False + return self.send_message_thread(message, mode, translate) + + @Slot(str, str, bool) + def send_message_thread(self, message, mode, translate=True): + try: + yes, no, ok = QMessageBox.Yes, QMessageBox.No, QMessageBox.Ok + select = None + + if not isinstance(message, str): + message = str(message) + if message and translate: + message = self.trans(self.pyas_config["language"], message) + + m = mode.lower() + if m == "quest": + select = QMessageBox.question(self, "Quest", message, yes | no) == yes + elif m == "info": + select = QMessageBox.information(self, "Info", message, ok) == ok + elif m == "error": + select = QMessageBox.critical(self, "Error", message, ok) == ok + elif m == "notify": + now = time.strftime("%Y-%m-%d %H:%M:%S") + self.tray_icon.showMessage(now, message, QSystemTrayIcon.Information, 5000) + elif m == "files": + select = QFileDialog.getOpenFileNames(self, message)[0] + elif m == "file": + select = QFileDialog.getOpenFileName(self, message)[0] + elif m == "path": + select = QFileDialog.getExistingDirectory(self, message) + elif m == "save": + select = QFileDialog.getSaveFileName(self, message)[0] + + now = time.strftime("%Y-%m-%d %H:%M:%S") + output = f"[{now}] | {mode} | {message} | {select}" + log_widget = self.widgets.get("log_text") + if log_widget: + log_widget.append(output) + return select or False + except Exception: + traceback.print_exc() + return False + + def invoke_method(self, obj, method, *args): + q_args = [] + for arg in args: + if isinstance(arg, int): + q_args.append(Q_ARG("int", arg)) + elif isinstance(arg, QSystemTrayIcon.MessageIcon): + q_args.append(Q_ARG("QSystemTrayIcon::MessageIcon", arg)) + else: + q_args.append(Q_ARG("QString", str(arg))) + QMetaObject.invokeMethod(obj, method, Qt.QueuedConnection, *q_args) + + def norm_path(self, path): + if isinstance(path, (list, tuple, set)): + t = type(path) + return t(self.norm_path(p) for p in path) + if isinstance(path, str): + return os.path.normpath(os.path.abspath(path)) + return path + +#################################################################################################### + + @Slot() + def slot_scan_reset(self): + self.widgets["virus_list"].clear() + self.widgets["solve_button"].hide() + self.widgets["stop_button"].show() + self.widgets["method_button"].hide() + self.widgets["method_window"].hide() + + @Slot(str) + def slot_progress_title(self, text: str): + self.widgets["progress_title"].setText(text) + + @Slot(str) + def slot_scan_progress(self, text): + self.widgets["progress_text"].setText(text) + + @Slot(str) + def slot_scan_add_virus(self, item_text): + item = QListWidgetItem(item_text) + item.setFlags(item.flags() | Qt.ItemIsUserCheckable) + item.setCheckState(Qt.Checked) + self.widgets["virus_list"].addItem(item) + + @Slot(str) + def slot_scan_result(self, msg): + self.widgets["progress_text"].setText(msg) + count = len(self.virus_results) + self.widgets["solve_button"].setVisible(count > 0) + self.widgets["stop_button"].hide() + self.widgets["method_button"].show() + self.send_message(msg, "notify", True) + +#################################################################################################### + + def init_scan(self): + try: + self.scan_running = True + self.virus_results = [] + self.scan_count = 0 + self.scan_start = time.time() + self.last_widget = self.widgets.get("scan_window") + self.scan_reset_signal.emit() + self.progress_title_signal.emit(self.trans(self.pyas_config["language"], "正在掃描")) + self.scan_progress_signal.emit(self.trans(self.pyas_config["language"], "正在初始化中")) + except Exception as e: + self.send_message(e, "warn", False) + + def file_button(self): + targets = self.send_message("檔案掃描", "files", True) + if targets: + self.init_scan() + self.start_daemon_thread(self.scan_worker, targets) + + def path_button(self): + targets = self.send_message("路徑掃描", "path", True) + if targets: + self.init_scan() + self.start_daemon_thread(self.scan_worker, targets) + + def full_button(self): + targets = [f"{chr(d)}:/" for d in range(65, 91) if os.path.exists(f"{chr(d)}:/")] + if targets: + self.init_scan() + self.start_daemon_thread(self.scan_worker, targets) + +#################################################################################################### + + def scan_worker(self, targets): + try: + for file_path in self.yield_files(targets): + if not self.scan_running: + break + try: + norm_path = self.norm_path(file_path) + if self.is_in_whitelist(norm_path): + continue + self.scan_count += 1 + state = self.scan_engine(norm_path) + if state: + self.virus_results.append(norm_path) + self.scan_add_virus_signal.emit(f"[{state}] > {norm_path}") + self.scan_progress_signal.emit(norm_path) + except Exception as e: + self.send_message(e, "warn", False) + finally: + elapsed = int(time.time() - self.scan_start) + count = len(self.virus_results) + result = ( + f"發現 {count} 個病毒，耗時 {elapsed} 秒，共掃描 {self.scan_count} 檔案" + if count else f"未發現病毒，耗時 {elapsed} 秒，共掃描 {self.scan_count} 檔案") + self.progress_title_signal.emit(self.trans(self.pyas_config["language"], "病毒掃描")) + self.scan_result_signal.emit(self.trans(self.pyas_config["language"], result)) + + def yield_files(self, targets): + if isinstance(targets, str): + if os.path.isdir(targets): + for root, _, files in os.walk(targets): + for f in files: + yield self.norm_path(os.path.join(root, f)) + elif os.path.isfile(targets): + yield self.norm_path(targets) + elif isinstance(targets, (list, tuple, set)): + for t in targets: + yield from self.yield_files(t) + + def scan_engine(self, file_path): + try: + label, level = self.model.model_scan(file_path) + if label and level >= self.pyas_config.get("sensitive", 95): + return f"{label}.{level}" + + if self.pyas_config.get("extension_switch", False): + label, level = self.rule.yara_scan(file_path) + if label and level: + return f"{label}.{level}" + except Exception as e: + self.send_message(e, "warn", False) + return None + +#################################################################################################### + + def solve_button(self): + deleted = 0 + virus_list = self.widgets["virus_list"] + for i in reversed(range(virus_list.count())): + item = virus_list.item(i) + if item.checkState() == Qt.Checked: + try: + file_path = self.norm_path(item.text().split(">")[1].strip()) + os.remove(file_path) + deleted += 1 + except Exception as e: + self.send_message(e, "warn", False) + virus_list.takeItem(i) + self.widgets["solve_button"].hide() + msg = f"已清理 {deleted} 個病毒檔案" if deleted else "無檔案被刪除" + self.widgets["progress_text"].setText(self.trans(self.pyas_config["language"], msg)) + + def stop_button(self): + self.scan_running = False + self.widgets["stop_button"].hide() + +#################################################################################################### + + def list_process(self): + pe = self.get_process_entry() + snapshot = self.kernel32.CreateToolhelp32Snapshot(0x00000002, 0) + result, success = [], self.kernel32.Process32First(snapshot, ctypes.byref(pe)) + while success: + pid = pe.th32ProcessID + name, file_path = (None, None) + if pid > 4: + name, file_path = self.get_exe_info(pid) + if not name: + raw = bytes(pe.szExeFile).split(b"\x00", 1)[0] + if raw: + try: + name = raw.decode("mbcs", errors="replace") or None + except Exception: + name = None + result.append((pid, name, file_path)) + success = self.kernel32.Process32Next(snapshot, ctypes.byref(pe)) + self.kernel32.CloseHandle(snapshot) + return result + + def refresh_process(self): + procs = self.list_process() + self.proc_pid_map = [pid for pid, _, _ in procs] + widget = self.widgets.get("manage_list") + info_text = self.widgets.get("info_text") + if widget: + model = QStringListModel([f"{name or 'None'} | {pid} | {file_path or 'None'}" for pid, name, file_path in procs]) + widget.setModel(model) + if info_text: + info_text.setText(str(len(self.proc_pid_map))) + + def device_path_to_drive(self, path): + if not path: + return "" + for d in range(65, 91): + drive = f"{chr(d)}:" + buf = ctypes.create_unicode_buffer(1024) + if self.kernel32.QueryDosDeviceW(drive, buf, 1024): + dev = buf.value + if path.startswith(dev): + return path.replace(dev, drive, 1) + return path + + def start_process_timer(self): + if hasattr(self, "process_timer") and self.process_timer.isActive(): + return + self.process_timer = QTimer(self) + self.process_timer.timeout.connect(self.refresh_process) + self.process_timer.start(0) + + def stop_process_timer(self): + if hasattr(self, "process_timer"): + self.process_timer.stop() + + def setup_process_menu(self): + widget = self.widgets.get("manage_list") + if widget: + widget.setContextMenuPolicy(Qt.CustomContextMenu) + widget.customContextMenuRequested.connect(self.show_process_menu) + + def show_process_menu(self, pos): + widget = self.widgets.get("manage_list") + if not widget: + return + idx = widget.indexAt(pos) + if not idx.isValid() or idx.row() >= len(self.proc_pid_map): + return + pid = self.proc_pid_map[idx.row()] + menu = QMenu() + kill_action = menu.addAction("結束進程") + if menu.exec(widget.viewport().mapToGlobal(pos)) == kill_action: + self.kill_process(pid) + + def kill_process(self, pid): + try: + h = self.kernel32.OpenProcess(0x1F0FFF, False, pid) + if h: + file_path = self.norm_path(self.get_process_file(h)) + if hasattr(self, "file_pyas") and file_path == getattr(self, "file_pyas", None): + self.close() + else: + self.kernel32.TerminateProcess(h, 0) + self.kernel32.CloseHandle(h) + except Exception as e: + self.send_message(e, "warn", False) + self.refresh_process() + +#################################################################################################### + + def clean_button(self): + try: + if self.send_message("您確定要清理系統垃圾嗎?", "quest", True): + self.total_deleted_size = 0 + for path in [self.path_temp, self.path_systemp]: + self.traverse_temp(path) + size_text = self.format_size(self.total_deleted_size) + self.send_message(f"成功清理了 {size_text} 系統垃圾", "info", True) + except Exception as e: + self.send_message(e, "warn", False) + + def traverse_temp(self, path): + try: + if not os.path.exists(path): + return + for fd in os.listdir(path): + file = os.path.join(path, fd) + try: + QApplication.processEvents() + if os.path.isdir(file): + self.traverse_temp(file) + try: + os.rmdir(file) + except Exception: + pass + else: + file_size = os.path.getsize(file) + os.remove(file) + self.total_deleted_size += file_size + except Exception: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def format_size(self, size): + units = ["B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB"] + for unit in units: + if size < 1024 or unit == units[-1]: + return f"{size:.2f} {unit}" + size /= 1024 + +#################################################################################################### + + def repair_button(self): + try: + if self.send_message("您確定要修復系統檔案嗎?", "quest", True): + self.repair_system_mbr() + self.repair_system_wallpaper() + self.repair_system_restrict() + self.repair_system_file_type() + self.repair_system_file_icon() + self.repair_system_image() + self.send_message("修復系統檔案成功", "info", True) + except Exception as e: + self.send_message(e, "warn", False) + + def backup_mbr(self, max_drives=26): + self.mbr_backup = {} + for drive in range(max_drives): + drive_path = rf"\\.\PhysicalDrive{drive}" + try: + with open(drive_path, "rb") as f: + mbr = f.read(512) + if mbr[510:512] == b"\x55\xAA": + self.mbr_backup[drive] = mbr + except FileNotFoundError: + continue + except PermissionError as e: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_mbr(self): + if not hasattr(self, "mbr_backup") or not self.mbr_backup: + return + for drive, mbr_value in self.mbr_backup.items(): + drive_path = rf"\\.\PhysicalDrive{drive}" + try: + with open(drive_path, "rb+") as f: + current = f.read(512) + if current != mbr_value: + f.seek(0) + f.write(mbr_value) + except PermissionError as e: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_wallpaper(self): + try: + wallpaper = os.path.join(self.path_system, "web", "wallpaper", "Windows", "img0.jpg") + if not os.path.exists(wallpaper): + return + key = r"Control Panel\\Desktop" + with winreg.OpenKey(winreg.HKEY_CURRENT_USER, key, 0, winreg.KEY_SET_VALUE) as reg: + winreg.SetValueEx(reg, "Wallpaper", 0, winreg.REG_SZ, wallpaper) + theme_dir = os.path.join(self.path_appdata, "Microsoft", "Windows", "Themes") + for fname in ["TranscodedWallpaper", "TranscodedWallpaper.tmp"]: + fpath = os.path.join(theme_dir, fname) + if os.path.exists(fpath): + try: + os.remove(fpath) + except Exception: + pass + cache_dir = os.path.join(theme_dir, "CachedFiles") + if os.path.exists(cache_dir): + try: + shutil.rmtree(cache_dir, ignore_errors=True) + except Exception: + pass + self.user32.SystemParametersInfoW(20, 0, wallpaper, 3) + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_restrict(self): + try: + permissions = [ + "NoControlPanel", "NoDrives", "NoFileMenu", "NoFind", "NoStartMenuPinnedList", + "NoSetFolders", "NoSetFolderOptions", "NoViewOnDrive", "NoClose", "NoDesktop", + "NoLogoff", "NoFolderOptions", "RestrictRun", "NoViewContextMenu", "HideClock", + "NoStartMenuMyGames", "NoStartMenuMyMusic", "DisableCMD", "NoAddingComponents", + "NoWinKeys", "NoStartMenuLogOff", "NoSimpleNetIDList", "NoLowDiskSpaceChecks", + "DisableLockWorkstation","Restrict_Run", "DisableTaskMgr", "DisableRegistryTools", + "DisableChangePassword", "Wallpaper", "NoComponents", "NoStartMenuMorePrograms", + "NoActiveDesktop", "NoSetActiveDesktop", "NoRecentDocsMenu", "NoWindowsUpdate", + "NoChangeStartMenu", "NoFavoritesMenu", "NoRecentDocsHistory", "NoSetTaskbar", + "NoSMHelp", "NoTrayContextMenu", "NoManageMyComputerVerb", "NoRealMode", "NoRun", + "ClearRecentDocsOnExit", "NoActiveDesktopChanges", "NoStartMenuNetworkPlaces" + ] + restrict_paths = [ + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Policies\Microsoft\MMC"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Policies\Microsoft\Windows\System"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\Explorer"), + (winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\Explorer"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Policies\System"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows NT\CurrentVersion\Winlogon"), + (winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Policies\Microsoft\Windows\System"), + ] + + for hkey, path in restrict_paths: + try: + with winreg.OpenKey(hkey, path, 0, winreg.KEY_SET_VALUE | winreg.KEY_WRITE) as reg: + for value in permissions: + try: + winreg.DeleteValue(reg, value) + except FileNotFoundError: + continue + except FileNotFoundError: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_file_type(self): + try: + file_types = [".exe", ".bat", ".cmd", ".com"] + for root in [winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER]: + for ext in file_types: + try: + with winreg.CreateKey(root, rf"SOFTWARE\Classes\{ext}") as reg: + winreg.SetValue(reg, "", winreg.REG_SZ, "exefile" if ext == ".exe" else ext[1:]+"file") + except Exception: + continue + + for shell_cmd in ["open", "runas"]: + try: + with winreg.CreateKey(root, r"SOFTWARE\Classes\exefile\shell\{}\command".format(shell_cmd)) as reg: + winreg.SetValue(reg, "", winreg.REG_SZ, '"%1" %*') + except Exception: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_file_icon(self): + try: + for root in [winreg.HKEY_LOCAL_MACHINE, winreg.HKEY_CURRENT_USER]: + try: + with winreg.CreateKey(root, r"SOFTWARE\Classes\exefile\DefaultIcon") as reg: + winreg.SetValue(reg, "", winreg.REG_SZ, "%1") + except Exception: + continue + except Exception as e: + self.send_message(e, "warn", False) + + def repair_system_image(self): + try: + key_path = r"SOFTWARE\Microsoft\Windows NT\CurrentVersion\Image File Execution Options" + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path, 0, winreg.KEY_ALL_ACCESS) as reg: + i = 0 + while True: + try: + subkey = winreg.EnumKey(reg, i) + subkey_path = key_path + "\\" + subkey + with winreg.OpenKey(reg, subkey, 0, winreg.KEY_ALL_ACCESS) as sub_reg: + for value in ["Debugger", "UseFilter", "GlobalFlag", "MitigationOptions"]: + try: + winreg.DeleteValue(sub_reg, value) + except FileNotFoundError: + continue + i += 1 + except OSError: + break + except Exception as e: + self.send_message(e, "warn", False) + +#################################################################################################### + + def popup_button(self): + self.config_list("block_list") + self.block_window = False + if self.send_message("請選擇要攔截的軟體彈窗", "quest", True): + while True: + QApplication.processEvents() + title, class_name, process_name = self.get_window_info(self.user32.GetForegroundWindow()) + if process_name and not any(self.window_rule_match(item, process_name, class_name, title) for item in self.pass_windows): + if self.send_message(f"您確定要攔截 {process_name} ({title}) 嗎?", "quest", True): + if self.add_window_rule(self.pyas_config["block_list"], process_name, class_name, title): + self.send_message(f"成功增加到彈窗攔截: {process_name} ({title})", "info", True) + else: + self.send_message("已存在彈窗攔截", "info", True) + break + self.start_daemon_thread(self.popup_intercept_thread) + + def popup_button_2(self): + self.config_list("block_list") + self.block_window = False + if self.send_message("請選擇要取消攔截的軟體彈窗", "quest", True): + while True: + QApplication.processEvents() + title, class_name, process_name = self.get_window_info(self.user32.GetForegroundWindow()) + if process_name and not any(self.window_rule_match(item, process_name, class_name, title) for item in self.pass_windows): + if self.send_message(f"您確定要取消攔截 {process_name} ({title}) 嗎?", "quest", True): + if self.remove_window_rule(self.pyas_config["block_list"], process_name, class_name, title): + self.send_message(f"成功取消彈窗攔截: {process_name} ({title})", "info", True) + else: + self.send_message("未找到彈窗攔截", "info", True) + break + self.start_daemon_thread(self.popup_intercept_thread) + + def window_rule_match(self, item, process_name, class_name, title): + for k, v in zip(["exe", "class", "title"], [process_name, class_name, title]): + if item.get(k, "") and item.get(k, "") != v: + return False + return True + + def add_window_rule(self, rule_list, process_name, class_name, title): + if not any(self.window_rule_match(item, process_name, class_name, title) for item in rule_list): + rule_list.append({"exe": process_name, "class": class_name, "title": title}) + self.save_config(self.file_config, self.pyas_config) + return True + return False + + def remove_window_rule(self, rule_list, process_name, class_name, title): + before = len(rule_list) + rule_list[:] = [item for item in rule_list if not self.window_rule_match(item, process_name, class_name, title)] + if len(rule_list) < before: + self.save_config(self.file_config, self.pyas_config) + return True + return False + + def popup_intercept_thread(self): + self.block_window = True + while self.block_window: + try: + time.sleep(0.2) + rules = self.pyas_config.get("block_list", []) + if not rules: + continue + for hWnd in self.get_all_windows(): + title, class_name, process_name = self.get_window_info(hWnd) + if (not any(self.window_rule_match(item, process_name, class_name, title) for item in self.pass_windows) and + any(self.window_rule_match(item, process_name, class_name, title) for item in rules)): + for msg in [0x0010, 0x0002, 0x0012, 0x0112]: + self.user32.SendMessageW(hWnd, msg, 0xF060, 0) + except Exception as e: + self.send_message(e, "warn", False) + +#################################################################################################### + + def get_all_windows(self): + self.hwnd_list = [] + WNDENUMPROC = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_int, ctypes.c_int) + self.user32.EnumWindows(WNDENUMPROC(self.enum_windows_callback), 0) + return self.hwnd_list + + def enum_windows_callback(self, hWnd, lParam): + self.hwnd_list.append(hWnd) + return True + + def get_window_info(self, hWnd): + length = self.user32.GetWindowTextLengthW(hWnd) + title = ctypes.create_unicode_buffer(length + 1) + self.user32.GetWindowTextW(hWnd, title, length + 1) + class_name = ctypes.create_unicode_buffer(256) + self.user32.GetClassNameW(hWnd, class_name, 256) + pid = ctypes.c_ulong() + self.user32.GetWindowThreadProcessId(hWnd, ctypes.byref(pid)) + process_name = "" + try: + process_name = self.get_process_name_by_pid(pid.value) + except Exception: + pass + return str(title.value), str(class_name.value), process_name + + def get_process_name_by_pid(self, pid): + snapshot = self.kernel32.CreateToolhelp32Snapshot(0x00000002, 0) + entry = PROCESSENTRY32() + entry.dwSize = ctypes.sizeof(PROCESSENTRY32) + name = "" + if self.kernel32.Process32First(snapshot, ctypes.byref(entry)): + while True: + if entry.th32ProcessID == pid: + name = entry.szExeFile.decode("mbcs").rstrip("\x00") + break + if not self.kernel32.Process32Next(snapshot, ctypes.byref(entry)): + break + self.kernel32.CloseHandle(snapshot) + return name + +#################################################################################################### + + def whitelist_button(self): + self.config_list("white_list") + files = self.send_message("選擇檔案", "files", True) + if files: + if self.send_message("您確定要增加到白名單嗎?", "quest"): + n = self.manage_named_list("white_list", files, action="add", with_hash=True) + self.send_message(f"成功增加到白名單，共 {n} 個檔案", "info") + + def whitelist_button_2(self): + self.config_list("white_list") + files = self.send_message("選擇檔案", "files", True) + if files: + if self.send_message("您確定要移除白名單嗎?", "quest"): + n = self.manage_named_list("white_list", files, action="remove", with_hash=True) + self.send_message(f"成功移除白名單，共 {n} 個檔案", "info") + + def quarantine_button(self): + self.config_list("quarantine") + files = self.send_message("選擇檔案", "files", True) + if files: + if self.send_message("您確定要增加到隔離區嗎?", "quest"): + n = self.manage_named_list("quarantine", files, action="add", with_hash=True, lock_func=self.lock_file) + if n > 0: + self.send_message(f"成功增加到隔離區，共 {n} 個檔案", "info") + + def quarantine_button_2(self): + self.config_list("quarantine") + quarantine_dir = os.path.join(self.path_config, "quarantine") + files = self.send_message("選擇檔案", "files", True) + if files: + if self.send_message("您確定要移除隔離區嗎?", "quest"): + n = self.manage_named_list("quarantine", files, action="remove", with_hash=True, lock_func=self.lock_file) + self.send_message(f"成功移除隔離區，共 {n} 個檔案", "info") + + def add_to_quarantine(self, files): + self.config_list("quarantine") + return self.manage_named_list("quarantine", files, action="add", with_hash=True, lock_func=self.lock_file) + + def is_in_whitelist(self, file_path): + norm_path = self.norm_path(file_path) + file_hash = self.calc_file_hash(norm_path) + return any(item["file"] == norm_path and item["hash"] == file_hash + for item in self.pyas_config.get("white_list", [])) + + def init_whitelist(self): + self.manage_named_list("white_list", [self.file_pyas], action="add", with_hash=True) + +#################################################################################################### + + def manage_named_list(self, list_key, files, action="add", with_hash=True, lock_func=None): + target_list = self.config_list(list_key) + files = files or [] + norm_paths = self.norm_path(files) + if isinstance(norm_paths, str): + norm_paths = [norm_paths] + + n = 0 + if action == "add": + for path in norm_paths: + QApplication.processEvents() + file_hash = self.calc_file_hash(path) if with_hash else "" + if not any(self.norm_path(item["file"]) == path and (not with_hash or item.get("hash", "") == file_hash) for item in target_list): + if lock_func: + try: + lock_func(path, True) + except Exception as e: + self.send_message(e, "warn", False) + target_list.append({"file": path, "hash": file_hash}) + n += 1 + + elif action == "remove": + remove_list = [item for item in target_list if item["file"] in norm_paths] + for item in remove_list: + try: + if lock_func: + lock_func(item["file"], False) + target_list.remove(item) + n += 1 + except Exception as e: + self.send_message(e, "warn", False) + if n: + self.save_config(self.file_config, self.pyas_config) + return n + + def calc_file_hash(self, file_path, block_size=65536): + try: + h = hashlib.sha256() + with open(file_path, "rb") as f: + for chunk in iter(lambda: f.read(block_size), b""): + h.update(chunk) + return h.hexdigest() + except Exception: + return "" + +#################################################################################################### + + def lock_file(self, file, lock): + try: + if lock: + if file not in self.virus_lock: + self.virus_lock[file] = os.open(file, os.O_RDWR) + msvcrt.locking(self.virus_lock[file], msvcrt.LK_NBRLCK, os.path.getsize(file)) + else: + if file in self.virus_lock: + msvcrt.locking(self.virus_lock[file], msvcrt.LK_UNLCK, os.path.getsize(file)) + os.close(self.virus_lock[file]) + del self.virus_lock[file] + except Exception as e: + self.send_message(e, "warn", False) + + def relock_file(self): + quarantine_list = self.pyas_config.get("quarantine", []) + for item in quarantine_list: + file = item["file"] + if os.path.exists(file): + try: + self.lock_file(file, True) + except Exception as e: + self.send_message(e, "warn", False) + +#################################################################################################### + + def protect_proc_thread(self): + self.exist_process = self.get_process_list() + while self.pyas_config.get("process_switch", False): + try: + time.sleep(0.2) + cur = self.get_process_list() + for pid in cur - self.exist_process: + self.handle_new_process(pid) + self.exist_process = cur + except Exception as e: + self.send_message(e, "warn", False) + + def get_process_list(self): + exist_process, pe = set(), self.get_process_entry() + hSnapshot = self.kernel32.CreateToolhelp32Snapshot(0x00000002, 0) + if self.kernel32.Process32First(hSnapshot, ctypes.byref(pe)): + while True: + exist_process.add(pe.th32ProcessID) + if not self.kernel32.Process32Next(hSnapshot, ctypes.byref(pe)): + break + self.kernel32.CloseHandle(hSnapshot) + return exist_process + + def handle_new_process(self, pid): + try: + h = self.kernel32.OpenProcess(0x1F0FFF, False, pid) + if not h: + return + file_path = self.norm_path(self.get_process_file(h)) + if file_path and os.path.exists(file_path) and not self.is_in_whitelist(file_path): + self.suspend_process(h, True) + state = self.scan_engine(file_path) + if state: + self.kernel32.TerminateProcess(h, 0) + self.send_message(f"進程防護攔截 | 靜態掃描攔截 | {pid} | {file_path} | None", "notify", True) + self.suspend_process(h, False) + self.kernel32.CloseHandle(h) + except Exception as e: + self.send_message(e, "warn", False) + + def suspend_process(self, h_process, suspend): + try: + if not hasattr(self.ntdll, "NtSuspendProcess"): + self.ntdll.NtSuspendProcess.argtypes = [ctypes.c_void_p] + self.ntdll.NtSuspendProcess.restype = ctypes.c_ulong + self.ntdll.NtResumeProcess.argtypes = [ctypes.c_void_p] + self.ntdll.NtResumeProcess.restype = ctypes.c_ulong + if suspend: + self.ntdll.NtSuspendProcess(h_process) + else: + self.ntdll.NtResumeProcess(h_process) + except Exception as e: + self.send_message(e, "warn", False) + + def get_process_file(self, h_process): + buf = ctypes.create_unicode_buffer(1024) + if self.psapi.GetProcessImageFileNameW(h_process, buf, 1024): + return self.norm_path(self.device_path_to_drive(buf.value)) + return "" + + def get_process_entry(self): + pe = PROCESSENTRY32() + pe.dwSize = ctypes.sizeof(PROCESSENTRY32) + return pe + + def get_exe_info(self, pid): + name, file_path = None, None + h = self.kernel32.OpenProcess(0x1000, False, pid) + if h: + buf = ctypes.create_unicode_buffer(1024) + if self.psapi.GetProcessImageFileNameW(h, buf, 1024): + path = self.norm_path(self.device_path_to_drive(buf.value)) + if path: + file_path = path + name = os.path.basename(path) + self.kernel32.CloseHandle(h) + return name, file_path + +#################################################################################################### + + def protect_file_thread(self): + FILE_LIST_DIRECTORY = 0x0001 + hDir = self.kernel32.CreateFileW(self.path_user, FILE_LIST_DIRECTORY, 0x00000007, None, 3, 0x02000000, None) + buffer = ctypes.create_string_buffer(65536) + + while self.pyas_config.get("document_switch", False): + try: + bytes_returned = ctypes.wintypes.DWORD() + res = self.kernel32.ReadDirectoryChangesW(hDir, buffer, ctypes.sizeof(buffer), True, + 0x0000001F, ctypes.byref(bytes_returned), None, None) + + if res: + notify = FILE_NOTIFY_INFORMATION.from_buffer_copy(buffer) + raw_filename = notify.FileName[:notify.FileNameLength // 2] + file_path = self.norm_path(os.path.join(self.path_user, raw_filename)) + + if notify.Action in [2, 3, 4] and not self.is_in_whitelist(file_path): + state = self.scan_engine(file_path) + if state: + self.add_to_quarantine([file_path]) + self.send_message(f"檔案防護攔截 | 靜態掃描攔截 | None | {file_path} | None", "notify", True) + except Exception as e: + self.send_message(e, "warn", False) + self.kernel32.CloseHandle(hDir) + +#################################################################################################### + + def protect_net_thread(self): + while self.pyas_config.get("network_switch", False): + try: + time.sleep(0.5) + conns = self.get_connections_list() + if not hasattr(self, "exist_connections") or self.exist_connections is None: + self.exist_connections = set() + if conns is None: + conns = set() + + for key in conns - self.exist_connections: + self.handle_new_connection(key) + self.exist_connections = conns + except Exception as e: + self.send_message(e, "warn", False) + self.exist_connections = set() + + def get_connections_list(self): + try: + connections = set() + size = ctypes.wintypes.DWORD() + AF_INET = 2 + TCP_TABLE_OWNER_PID_ALL = 5 + + ret = self.iphlpapi.GetExtendedTcpTable(None, ctypes.byref(size), True, AF_INET, TCP_TABLE_OWNER_PID_ALL, 0) + if ret != 122: + raise ctypes.WinError(ret) + + buf = ctypes.create_string_buffer(size.value) + ret = self.iphlpapi.GetExtendedTcpTable(buf, ctypes.byref(size), True, AF_INET, TCP_TABLE_OWNER_PID_ALL, 0) + if ret != 0: + raise ctypes.WinError(ret) + + num_entries = ctypes.cast(buf, ctypes.POINTER(ctypes.wintypes.DWORD)).contents.value + row_size = ctypes.sizeof(MIB_TCPROW_OWNER_PID) + offset = ctypes.sizeof(ctypes.wintypes.DWORD) + + for i in range(num_entries): + entry_addr = ctypes.addressof(buf) + offset + i * row_size + row = MIB_TCPROW_OWNER_PID.from_address(entry_addr) + connections.add((row.dwOwningPid, row.dwRemoteAddr, row.dwRemotePort)) + return connections + except Exception as e: + self.send_message(e, "warn", False) + return set() + + def handle_new_connection(self, key): + pid, remote_addr, remote_port = key + try: + h_process = self.kernel32.OpenProcess(0x1F0FFF, False, pid) + if not h_process: + return + file_path = self.norm_path(self.get_process_file(h_process)) + remote_ip = f"{remote_addr & 0xFF}.{(remote_addr >> 8) & 0xFF}.{(remote_addr >> 16) & 0xFF}.{(remote_addr >> 24) & 0xFF}" + + if file_path and os.path.exists(file_path) and not self.is_in_whitelist(file_path): + if hasattr(self.rule, "network") and remote_ip in self.rule.network: + self.kernel32.TerminateProcess(h_process, 0) + self.send_message(f"網路防護攔截 | 規則列表攔截 | {pid} | {file_path} | {remote_ip}", "notify", True) + self.kernel32.CloseHandle(h_process) + except Exception as e: + self.send_message(e, "warn", False) + +#################################################################################################### + + def protect_system_thread(self): + while self.pyas_config.get("system_switch", False): + try: + self.repair_system_mbr() + self.repair_system_restrict() + self.repair_system_file_type() + self.repair_system_file_icon() + self.repair_system_image() + time.sleep(0.2) + except Exception as e: + self.send_message(e, "warn", False) + +#################################################################################################### + + def install_system_driver(self): + service_name = "PYAS_Driver" + scm = self.advapi32.OpenSCManagerW(None, None, 0xF003F) + if not scm: + return False + + svc = self.advapi32.CreateServiceW(scm, service_name, service_name, 0xF01FF, + 0x00000001, 0x00000003, 0x00000001, self.path_drivers, None, None, None, None, None) + if not svc: + svc = self.advapi32.OpenServiceW(scm, service_name, 0xF01FF) + if not svc: + self.advapi32.CloseServiceHandle(scm) + return False + + res = self.advapi32.StartServiceW(svc,0,None) + self.advapi32.CloseServiceHandle(svc) + self.advapi32.CloseServiceHandle(scm) + return bool(res) + + def stop_system_driver(self): + service_name = "PYAS_Driver" + scm = self.advapi32.OpenSCManagerW(None, None, 0xF003F) + if not scm: + return False + + svc = self.advapi32.OpenServiceW(scm, service_name, 0xF01FF) + if not svc: + self.advapi32.CloseServiceHandle(scm) + return False + + status = SERVICE_STATUS() + self.advapi32.ControlService(svc, 0x00000001, ctypes.byref(status)) + self.advapi32.DeleteService(svc) + self.advapi32.CloseServiceHandle(svc) + self.advapi32.CloseServiceHandle(scm) + return True + +#################################################################################################### + + def pipe_server_thread(self): + # The source code cannot be made public due to security concerns + pass + +#################################################################################################### + +if __name__ == "__main__": + QGuiApplication.setHighDpiScaleFactorRoundingPolicy(Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) + app = QApplication(sys.argv) + MainWindow_Controller() + sys.exit(app.exec())