1405 lines
56 KiB
Python
1405 lines
56 KiB
Python
import os, re, io, csv, json, time, shutil, threading
|
|
import pefile, hashlib, winreg, requests, subprocess
|
|
import ctypes, ctypes.wintypes
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
PROCESS_TERMINATE = 0x0001
|
|
PROCESS_QUERY_INFORMATION = 0x0400
|
|
PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
|
|
PYAS_PROCESS_TERMINATE_ACCESS = PROCESS_TERMINATE | PROCESS_QUERY_INFORMATION | PROCESS_QUERY_LIMITED_INFORMATION
|
|
|
|
####################################################################################################
|
|
|
|
class PROCESSENTRY32W(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),
|
|
("pcPriClassBase", ctypes.wintypes.LONG),
|
|
("dwFlags", ctypes.wintypes.DWORD),
|
|
("szExeFile", ctypes.wintypes.WCHAR * 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 FILE_NOTIFY_INFORMATION(ctypes.Structure):
|
|
_fields_ = [
|
|
("NextEntryOffset", ctypes.wintypes.DWORD),
|
|
("Action", ctypes.wintypes.DWORD),
|
|
("FileNameLength", ctypes.wintypes.DWORD),
|
|
("FileName", ctypes.wintypes.WCHAR * 1024)
|
|
]
|
|
|
|
class PROCESS_BASIC_INFORMATION(ctypes.Structure):
|
|
_fields_ = [
|
|
("Reserved1", ctypes.wintypes.LPVOID),
|
|
("PebBaseAddress", ctypes.wintypes.LPVOID),
|
|
("Reserved2", ctypes.wintypes.LPVOID * 2),
|
|
("UniqueProcessId", ctypes.wintypes.LPVOID),
|
|
("Reserved3", ctypes.wintypes.LPVOID)
|
|
]
|
|
|
|
class SHQUERYRBINFO(ctypes.Structure):
|
|
_fields_ = [
|
|
("cbSize", ctypes.wintypes.DWORD),
|
|
("i64Size", ctypes.c_int64),
|
|
("i64NumItems", ctypes.c_int64)
|
|
]
|
|
|
|
class UNICODE_STRING(ctypes.Structure):
|
|
_fields_ = [
|
|
("Length", ctypes.wintypes.USHORT),
|
|
("MaximumLength", ctypes.wintypes.USHORT),
|
|
("Buffer", ctypes.c_void_p)
|
|
]
|
|
|
|
class FILTER_MESSAGE_HEADER(ctypes.Structure):
|
|
_fields_ = [
|
|
("ReplyLength", ctypes.wintypes.ULONG),
|
|
("MessageId", ctypes.c_uint64)
|
|
]
|
|
|
|
class PYAS_MESSAGE(ctypes.Structure):
|
|
_fields_ = [
|
|
("MessageCode", ctypes.wintypes.ULONG),
|
|
("ProcessId", ctypes.wintypes.ULONG),
|
|
("Path", ctypes.wintypes.WCHAR * 1024)
|
|
]
|
|
|
|
class PYAS_FULL_MESSAGE(ctypes.Structure):
|
|
_fields_ = [
|
|
("Header", FILTER_MESSAGE_HEADER),
|
|
("Data", PYAS_MESSAGE)
|
|
]
|
|
|
|
class PYAS_USER_MESSAGE(ctypes.Structure):
|
|
_fields_ = [
|
|
("Command", ctypes.wintypes.ULONG),
|
|
("Path", ctypes.wintypes.WCHAR * 1024)
|
|
]
|
|
|
|
class COPYDATASTRUCT(ctypes.Structure):
|
|
_fields_ = [
|
|
("dwData", ctypes.c_size_t),
|
|
("cbData", ctypes.wintypes.DWORD),
|
|
("lpData", ctypes.c_void_p)
|
|
]
|
|
|
|
class IO_COUNTERS(ctypes.Structure):
|
|
_fields_ = [
|
|
("ReadOperationCount", ctypes.c_ulonglong),
|
|
("WriteOperationCount", ctypes.c_ulonglong),
|
|
("OtherOperationCount", ctypes.c_ulonglong),
|
|
("ReadTransferCount", ctypes.c_ulonglong),
|
|
("WriteTransferCount", ctypes.c_ulonglong),
|
|
("OtherTransferCount", ctypes.c_ulonglong)
|
|
]
|
|
|
|
class MEMORY_BASIC_INFORMATION(ctypes.Structure):
|
|
_fields_ = [
|
|
("BaseAddress", ctypes.c_void_p),
|
|
("AllocationBase", ctypes.c_void_p),
|
|
("AllocationProtect", ctypes.wintypes.DWORD),
|
|
("RegionSize", ctypes.c_size_t),
|
|
("State", ctypes.wintypes.DWORD),
|
|
("Protect", ctypes.wintypes.DWORD),
|
|
("Type", ctypes.wintypes.DWORD)
|
|
]
|
|
|
|
class LUID(ctypes.Structure):
|
|
_fields_ = [
|
|
("LowPart", ctypes.wintypes.DWORD),
|
|
("HighPart", ctypes.wintypes.LONG)
|
|
]
|
|
|
|
class LUID_AND_ATTRIBUTES(ctypes.Structure):
|
|
_fields_ = [
|
|
("Luid", LUID),
|
|
("Attributes", ctypes.wintypes.DWORD)
|
|
]
|
|
|
|
class TOKEN_PRIVILEGES(ctypes.Structure):
|
|
_fields_ = [
|
|
("PrivilegeCount", ctypes.wintypes.DWORD),
|
|
("Privileges", LUID_AND_ATTRIBUTES * 1)
|
|
]
|
|
|
|
class SERVICE_STATUS_PROCESS(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),
|
|
("dwProcessId", ctypes.wintypes.DWORD),
|
|
("dwServiceFlags", ctypes.wintypes.DWORD)
|
|
]
|
|
|
|
class POINT(ctypes.Structure):
|
|
_fields_ = [
|
|
("x", ctypes.c_long),
|
|
("y", ctypes.c_long)
|
|
]
|
|
|
|
class RECT(ctypes.Structure):
|
|
_fields_ = [
|
|
("left", ctypes.c_long),
|
|
("top", ctypes.c_long),
|
|
("right", ctypes.c_long),
|
|
("bottom", ctypes.c_long)
|
|
]
|
|
|
|
####################################################################################################
|
|
|
|
class ToolsMixin:
|
|
def _find_windows_tool(self, *names):
|
|
windows_dir = os.environ.get("SystemRoot") or os.environ.get("WINDIR") or r"C:\Windows"
|
|
search_dirs = [
|
|
os.path.join(windows_dir, "Sysnative"),
|
|
os.path.join(windows_dir, "System32"),
|
|
windows_dir
|
|
]
|
|
|
|
for name in names:
|
|
executable_name = name if name.lower().endswith(".exe") else name + ".exe"
|
|
for directory in search_dirs:
|
|
candidate = os.path.join(directory, executable_name)
|
|
if os.path.isfile(candidate):
|
|
return candidate
|
|
|
|
candidate = shutil.which(name)
|
|
if candidate:
|
|
return candidate
|
|
|
|
return None
|
|
|
|
def _run_windows_tool(self, tool_names, arguments, **kwargs):
|
|
executable = self._find_windows_tool(*tool_names)
|
|
if not executable:
|
|
return None
|
|
|
|
options = {"creationflags": 0x08000000}
|
|
options.update(kwargs)
|
|
try:
|
|
return subprocess.run([executable, *arguments], **options)
|
|
except (FileNotFoundError, OSError):
|
|
return None
|
|
|
|
def _run_powershell(self, command, **kwargs):
|
|
return self._run_windows_tool(
|
|
("powershell.exe", "pwsh.exe"),
|
|
["-NoProfile", "-NonInteractive", "-ExecutionPolicy", "Bypass", "-WindowStyle", "Hidden", "-Command", command],
|
|
**kwargs
|
|
)
|
|
|
|
def _set_registry_autostart(self, enable):
|
|
run_key = r"Software\Microsoft\Windows\CurrentVersion\Run"
|
|
value_name = "PYAS_Security"
|
|
if enable:
|
|
command = f'"{self.file_pyas}" -hide'
|
|
return self._reg_write(winreg.HKEY_CURRENT_USER, run_key, value_name, winreg.REG_SZ, command)
|
|
|
|
try:
|
|
with winreg.OpenKey(winreg.HKEY_CURRENT_USER, run_key, 0, winreg.KEY_SET_VALUE) as reg:
|
|
try:
|
|
winreg.DeleteValue(reg, value_name)
|
|
except FileNotFoundError:
|
|
pass
|
|
return True
|
|
except FileNotFoundError:
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def _clear_event_log(self, log_name):
|
|
result = self._run_windows_tool(
|
|
("wevtutil.exe",),
|
|
["cl", log_name],
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL
|
|
)
|
|
if result is not None and result.returncode == 0:
|
|
return True
|
|
|
|
try:
|
|
wevtapi = ctypes.WinDLL("wevtapi", use_last_error=True)
|
|
wevtapi.EvtClearLog.argtypes = [ctypes.wintypes.HANDLE, ctypes.wintypes.LPCWSTR, ctypes.wintypes.LPCWSTR, ctypes.wintypes.DWORD]
|
|
wevtapi.EvtClearLog.restype = ctypes.wintypes.BOOL
|
|
return bool(wevtapi.EvtClearLog(None, log_name, None, 0))
|
|
except Exception:
|
|
return False
|
|
|
|
def _manage_service_native(self, service_name, action):
|
|
SC_MANAGER_CONNECT = 0x0001
|
|
SERVICE_CHANGE_CONFIG = 0x0002
|
|
DELETE = 0x00010000
|
|
SERVICE_NO_CHANGE = 0xFFFFFFFF
|
|
SERVICE_AUTO_START = 0x00000002
|
|
SERVICE_DISABLED = 0x00000004
|
|
|
|
try:
|
|
advapi32 = ctypes.WinDLL("advapi32", use_last_error=True)
|
|
advapi32.OpenSCManagerW.argtypes = [ctypes.wintypes.LPCWSTR, ctypes.wintypes.LPCWSTR, ctypes.wintypes.DWORD]
|
|
advapi32.OpenSCManagerW.restype = ctypes.wintypes.HANDLE
|
|
advapi32.OpenServiceW.argtypes = [ctypes.wintypes.HANDLE, ctypes.wintypes.LPCWSTR, ctypes.wintypes.DWORD]
|
|
advapi32.OpenServiceW.restype = ctypes.wintypes.HANDLE
|
|
advapi32.CloseServiceHandle.argtypes = [ctypes.wintypes.HANDLE]
|
|
advapi32.CloseServiceHandle.restype = ctypes.wintypes.BOOL
|
|
|
|
manager = advapi32.OpenSCManagerW(None, None, SC_MANAGER_CONNECT)
|
|
if not manager:
|
|
return False
|
|
|
|
service = None
|
|
try:
|
|
desired_access = DELETE if action == "delete" else SERVICE_CHANGE_CONFIG
|
|
service = advapi32.OpenServiceW(manager, service_name, desired_access)
|
|
if not service:
|
|
return False
|
|
|
|
if action == "delete":
|
|
advapi32.DeleteService.argtypes = [ctypes.wintypes.HANDLE]
|
|
advapi32.DeleteService.restype = ctypes.wintypes.BOOL
|
|
return bool(advapi32.DeleteService(service))
|
|
|
|
start_type = SERVICE_AUTO_START if action == "enable" else SERVICE_DISABLED
|
|
advapi32.ChangeServiceConfigW.argtypes = [
|
|
ctypes.wintypes.HANDLE,
|
|
ctypes.wintypes.DWORD,
|
|
ctypes.wintypes.DWORD,
|
|
ctypes.wintypes.DWORD,
|
|
ctypes.wintypes.LPCWSTR,
|
|
ctypes.wintypes.LPCWSTR,
|
|
ctypes.POINTER(ctypes.wintypes.DWORD),
|
|
ctypes.wintypes.LPCWSTR,
|
|
ctypes.wintypes.LPCWSTR,
|
|
ctypes.wintypes.LPCWSTR,
|
|
ctypes.wintypes.LPCWSTR
|
|
]
|
|
advapi32.ChangeServiceConfigW.restype = ctypes.wintypes.BOOL
|
|
return bool(advapi32.ChangeServiceConfigW(
|
|
service,
|
|
SERVICE_NO_CHANGE,
|
|
start_type,
|
|
SERVICE_NO_CHANGE,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
None,
|
|
None
|
|
))
|
|
finally:
|
|
if service:
|
|
advapi32.CloseServiceHandle(service)
|
|
advapi32.CloseServiceHandle(manager)
|
|
except Exception:
|
|
return False
|
|
|
|
def _manage_scheduled_task(self, task_path, action):
|
|
schtasks_action = "/delete" if action == "delete" else "/change"
|
|
arguments = [schtasks_action, "/tn", task_path]
|
|
if action == "delete":
|
|
arguments.append("/f")
|
|
else:
|
|
arguments.append("/enable" if action == "enable" else "/disable")
|
|
|
|
result = self._run_windows_tool(
|
|
("schtasks.exe",),
|
|
arguments,
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
if result is not None and result.returncode == 0:
|
|
return True
|
|
|
|
escaped_path = task_path.replace("'", "''")
|
|
operation = {
|
|
"delete": "Unregister-ScheduledTask -InputObject $Task -Confirm:$false",
|
|
"enable": "Enable-ScheduledTask -InputObject $Task | Out-Null",
|
|
"disable": "Disable-ScheduledTask -InputObject $Task | Out-Null"
|
|
}.get(action)
|
|
if not operation:
|
|
return False
|
|
|
|
command = (
|
|
f"$FullName = '{escaped_path}'; "
|
|
"$Task = Get-ScheduledTask -ErrorAction SilentlyContinue | "
|
|
"Where-Object { ($_.TaskPath + $_.TaskName) -eq $FullName } | Select-Object -First 1; "
|
|
f"if ($Task) {{ {operation} }} else {{ exit 1 }}"
|
|
)
|
|
result = self._run_powershell(command, capture_output=True, text=True)
|
|
return bool(result and result.returncode == 0)
|
|
|
|
def _reg_read(self, root, path, value_name):
|
|
try:
|
|
with winreg.OpenKey(root, path, 0, winreg.KEY_READ) as reg:
|
|
val, _ = winreg.QueryValueEx(reg, value_name)
|
|
return val
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
def _reg_write(self, root, path, value_name, value_type, value):
|
|
try:
|
|
with winreg.CreateKey(root, path) as reg:
|
|
if value_name is None:
|
|
winreg.SetValue(reg, "", value_type, value)
|
|
else:
|
|
winreg.SetValueEx(reg, value_name, 0, value_type, value)
|
|
return True
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
def _reg_delete(self, root, path, value_name=None):
|
|
try:
|
|
if value_name is None:
|
|
winreg.DeleteKey(root, path)
|
|
else:
|
|
with winreg.OpenKey(root, path, 0, winreg.KEY_SET_VALUE | winreg.KEY_WRITE) as reg:
|
|
winreg.DeleteValue(reg, value_name)
|
|
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
####################################################################################################
|
|
|
|
def manage_autostart(self, enable):
|
|
task_name = "PYAS_Security_ATS"
|
|
executable = os.path.normpath(self.file_pyas)
|
|
escaped_executable = executable.replace("'", "''")
|
|
errors = []
|
|
|
|
if enable:
|
|
task_command = f'"{executable}" -hide'
|
|
result = self._run_windows_tool(
|
|
("schtasks.exe",),
|
|
["/Create", "/TN", task_name, "/TR", task_command, "/SC", "ONLOGON", "/RL", "HIGHEST", "/F"],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
if result and result.returncode == 0:
|
|
self._set_registry_autostart(False)
|
|
return True
|
|
if result:
|
|
errors.append((result.stderr or result.stdout or f"schtasks exit {result.returncode}").strip())
|
|
|
|
command = (
|
|
f"$Action = New-ScheduledTaskAction -Execute '{escaped_executable}' -Argument '-hide'; "
|
|
"$Trigger = New-ScheduledTaskTrigger -AtLogOn; "
|
|
"$Settings = New-ScheduledTaskSettingsSet -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries; "
|
|
f"Register-ScheduledTask -TaskName '{task_name}' -Action $Action -Trigger $Trigger -Settings $Settings -RunLevel Highest -Force | Out-Null"
|
|
)
|
|
result = self._run_powershell(command, capture_output=True, text=True)
|
|
if result and result.returncode == 0:
|
|
self._set_registry_autostart(False)
|
|
return True
|
|
if result:
|
|
errors.append((result.stderr or result.stdout or f"PowerShell exit {result.returncode}").strip())
|
|
|
|
if self._set_registry_autostart(True):
|
|
self.write_log("INFO", "manage_autostart", detail="Scheduled Task unavailable; using HKCU Run fallback", success=True)
|
|
return True
|
|
else:
|
|
task_removed = False
|
|
query_result = self._run_windows_tool(
|
|
("schtasks.exe",),
|
|
["/Query", "/TN", task_name],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
result = None
|
|
if query_result is not None and query_result.returncode != 0:
|
|
task_removed = True
|
|
elif query_result is not None:
|
|
result = self._run_windows_tool(
|
|
("schtasks.exe",),
|
|
["/Delete", "/TN", task_name, "/F"],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
task_removed = bool(result and result.returncode == 0)
|
|
|
|
if query_result is None or not task_removed:
|
|
command = (
|
|
f"$Task = Get-ScheduledTask -TaskName '{task_name}' -ErrorAction SilentlyContinue; "
|
|
"if ($Task) { $Task | Unregister-ScheduledTask -Confirm:$false }"
|
|
)
|
|
powershell_result = self._run_powershell(command, capture_output=True, text=True)
|
|
if powershell_result is not None:
|
|
task_removed = powershell_result.returncode == 0
|
|
result = powershell_result
|
|
|
|
if result and result.returncode != 0:
|
|
errors.append((result.stderr or result.stdout or f"task removal exit {result.returncode}").strip())
|
|
|
|
registry_removed = self._set_registry_autostart(False)
|
|
if registry_removed and (task_removed or query_result is None and result is None):
|
|
return True
|
|
|
|
detail = "; ".join(error for error in errors if error) or "No supported autostart method was available"
|
|
self.write_log("WARN", "manage_autostart", detail=detail, success=False)
|
|
return False
|
|
|
|
####################################################################################################
|
|
|
|
def manage_named_list(self, list_key, files, action="add", lock_func=None):
|
|
if list_key == "quarantine" and lock_func is None:
|
|
lock_func = self.lock_file
|
|
|
|
norm_paths = self.norm_path(files or [], must_exist=False)
|
|
if isinstance(norm_paths, str):
|
|
norm_paths = [norm_paths] if norm_paths else []
|
|
|
|
if not norm_paths:
|
|
return 0
|
|
|
|
if list_key == "custom_rule":
|
|
acted_items = []
|
|
os.makedirs(self.path_rules, exist_ok=True)
|
|
|
|
for file_path in norm_paths:
|
|
if action == "add":
|
|
if os.path.exists(file_path) and file_path.lower().endswith(".json"):
|
|
dest = os.path.join(self.path_rules, os.path.basename(file_path))
|
|
|
|
if file_path != dest:
|
|
try:
|
|
shutil.copy2(file_path, dest)
|
|
acted_items.append(dest)
|
|
|
|
except Exception:
|
|
pass
|
|
else:
|
|
acted_items.append(dest)
|
|
|
|
elif action == "remove":
|
|
try:
|
|
if os.path.exists(file_path):
|
|
os.remove(file_path)
|
|
|
|
acted_items.append(file_path)
|
|
except Exception:
|
|
pass
|
|
|
|
if acted_items and getattr(self, 'driver_port', None):
|
|
self.clear_driver_rules()
|
|
for f in os.listdir(self.path_rules):
|
|
if f.lower().endswith(".json"):
|
|
self.load_driver_rule_file(os.path.join(self.path_rules, f))
|
|
|
|
return len(acted_items)
|
|
|
|
if action == "add":
|
|
if list_key == "white_list":
|
|
self.remove_list_items("quarantine", norm_paths)
|
|
|
|
elif list_key == "quarantine":
|
|
self.remove_list_items("white_list", norm_paths)
|
|
|
|
acted_items = []
|
|
with self.lock_config:
|
|
target_list = self.pyas_config.setdefault(list_key, [])
|
|
if action == "add":
|
|
for path in norm_paths:
|
|
path_case = os.path.normcase(path)
|
|
exists = False
|
|
|
|
for item in target_list:
|
|
val = item.get("file", "") if isinstance(item, dict) else item
|
|
np = self.norm_path(val, must_exist=False)
|
|
|
|
if np and os.path.normcase(np) == path_case:
|
|
exists = True
|
|
break
|
|
|
|
if not exists:
|
|
if lock_func:
|
|
lock_func(path, True)
|
|
|
|
new_item = {"file": path, "time": time.time()}
|
|
is_directory = None
|
|
if list_key == "white_list":
|
|
is_directory = os.path.isdir(path)
|
|
new_item["is_dir"] = is_directory
|
|
|
|
target_list.append(new_item)
|
|
acted_items.append(path)
|
|
if list_key == "white_list":
|
|
self.sync_driver_whitelist(path, True, is_directory)
|
|
|
|
elif action == "remove":
|
|
norm_paths_case = {os.path.normcase(p) for p in norm_paths}
|
|
new_list = []
|
|
|
|
for item in target_list:
|
|
val = item.get("file", "") if isinstance(item, dict) else item
|
|
if val:
|
|
np = self.norm_path(val, must_exist=False)
|
|
if np and os.path.normcase(np) in norm_paths_case:
|
|
if lock_func:
|
|
lock_func(val, False)
|
|
|
|
acted_items.append(val)
|
|
if list_key == "white_list":
|
|
is_directory = item.get("is_dir") if isinstance(item, dict) else None
|
|
self.sync_driver_whitelist(val, False, is_directory)
|
|
|
|
continue
|
|
new_list.append(item)
|
|
target_list[:] = new_list
|
|
|
|
if acted_items:
|
|
self.write_log("INFO", "Config Update", detail=f"List [{list_key}] {action}: {acted_items}")
|
|
self.save_config()
|
|
|
|
return len(acted_items)
|
|
|
|
####################################################################################################
|
|
|
|
def remove_list_items(self, list_key, paths_to_remove):
|
|
if list_key == "custom_rule":
|
|
return self.manage_named_list(list_key, paths_to_remove, action="remove") > 0
|
|
|
|
with self.lock_config:
|
|
target_list = self.pyas_config.get(list_key, [])
|
|
original_len = len(target_list)
|
|
|
|
norm_paths_to_remove = set()
|
|
for p in paths_to_remove:
|
|
np = self.norm_path(p, must_exist=False)
|
|
if np:
|
|
norm_paths_to_remove.add(os.path.normcase(np))
|
|
|
|
if list_key == "quarantine":
|
|
for item in target_list:
|
|
val = item.get("file") if isinstance(item, dict) else item
|
|
if val:
|
|
np = self.norm_path(val, must_exist=False)
|
|
if np and os.path.normcase(np) in norm_paths_to_remove:
|
|
self.lock_file(val, False)
|
|
|
|
new_list = []
|
|
removed_items = []
|
|
for item in target_list:
|
|
val = item.get("file") or item.get("exe") or item.get("title") if isinstance(item, dict) else item
|
|
if val:
|
|
np = self.norm_path(val, must_exist=False)
|
|
if np and os.path.normcase(np) in norm_paths_to_remove:
|
|
removed_items.append(val)
|
|
if list_key == "white_list":
|
|
is_directory = item.get("is_dir") if isinstance(item, dict) else None
|
|
self.sync_driver_whitelist(val, False, is_directory)
|
|
|
|
continue
|
|
new_list.append(item)
|
|
|
|
self.pyas_config[list_key] = new_list
|
|
|
|
if len(self.pyas_config[list_key]) < original_len:
|
|
self.write_log("INFO", "Config Update", detail=f"List [{list_key}] remove: {removed_items}")
|
|
self.save_config()
|
|
return True
|
|
|
|
return False
|
|
|
|
####################################################################################################
|
|
|
|
def extract_list_items(self, paths, dest_dir):
|
|
if not dest_dir or not os.path.exists(dest_dir):
|
|
return False
|
|
|
|
extracted_count = 0
|
|
for raw_path in paths:
|
|
src = self.norm_path(raw_path, must_exist=True)
|
|
if not src:
|
|
continue
|
|
|
|
base_name = os.path.basename(src)
|
|
target_path = os.path.join(dest_dir, base_name)
|
|
|
|
if os.path.exists(target_path):
|
|
name, ext = os.path.splitext(base_name)
|
|
counter = 1
|
|
|
|
while os.path.exists(target_path):
|
|
target_path = os.path.join(dest_dir, f"{name} ({counter}){ext}")
|
|
counter += 1
|
|
|
|
was_locked = False
|
|
src_norm = os.path.normcase(src)
|
|
src_dir = src_norm + os.sep
|
|
|
|
with self.lock_file_ops:
|
|
for locked_path in self.virus_lock:
|
|
locked_norm = os.path.normcase(locked_path)
|
|
if locked_norm == src_norm or locked_norm.startswith(src_dir):
|
|
was_locked = True
|
|
break
|
|
|
|
if was_locked:
|
|
self.lock_file(src, False)
|
|
|
|
try:
|
|
if os.path.isdir(src):
|
|
shutil.copytree(src, target_path)
|
|
else:
|
|
shutil.copy2(src, target_path)
|
|
extracted_count += 1
|
|
self.write_log("INFO", "File Extract", source=src, target=target_path, operate=True)
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "extract_list_items", source=src, detail=str(e), success=False)
|
|
finally:
|
|
if was_locked:
|
|
self.lock_file(src, True)
|
|
|
|
return extracted_count > 0
|
|
|
|
####################################################################################################
|
|
|
|
def start_daemon_thread(self, target, *args, **kwargs):
|
|
t = threading.Thread(target=target, args=args, kwargs=kwargs, daemon=True)
|
|
t.start()
|
|
return t
|
|
|
|
def norm_path(self, path, must_exist=True):
|
|
if isinstance(path, list):
|
|
return [p for p in (self.norm_path(x, must_exist) for x in path) if p]
|
|
|
|
if isinstance(path, str):
|
|
try:
|
|
ap = os.path.normpath(os.path.abspath(path))
|
|
return ap if (not must_exist or os.path.exists(ap)) else None
|
|
|
|
except Exception:
|
|
return None
|
|
|
|
return path
|
|
|
|
def path_equal(self, a, b):
|
|
pa, pb = self.norm_path(a, must_exist=False), self.norm_path(b, must_exist=False)
|
|
return os.path.normcase(pa) == os.path.normcase(pb) if pa and pb else False
|
|
|
|
####################################################################################################
|
|
|
|
def get_file_version(self, file_path):
|
|
try:
|
|
pe = pefile.PE(file_path, fast_load=True)
|
|
pe.parse_data_directories(directories=[pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_RESOURCE']])
|
|
|
|
if hasattr(pe, 'FileInfo'):
|
|
for fl in pe.FileInfo:
|
|
for info in fl:
|
|
if getattr(info, 'name', '') in ('StringFileInfo', b'StringFileInfo'):
|
|
|
|
for st in getattr(info, 'StringTable', []):
|
|
for k, val in st.entries.items():
|
|
k_str = k.decode('utf-8', 'ignore') if isinstance(k, bytes) else str(k)
|
|
v_str = val.decode('utf-8', 'ignore') if isinstance(val, bytes) else str(val)
|
|
if k_str == 'FileVersion':
|
|
pe.close()
|
|
return v_str.strip()
|
|
|
|
pe.close()
|
|
except Exception:
|
|
pass
|
|
|
|
return "0.0.0.0"
|
|
|
|
def compare_versions(self, v1, v2):
|
|
try:
|
|
val1 = tuple(int(x) for x in re.findall(r"\d+", re.sub(r"^[vV]\s*", "", str(v1))))
|
|
val2 = tuple(int(x) for x in re.findall(r"\d+", re.sub(r"^[vV]\s*", "", str(v2))))
|
|
return val1 >= val2
|
|
|
|
except Exception:
|
|
return False
|
|
|
|
def check_update(self):
|
|
try:
|
|
current = self.pyas_config.get("version", "0.0.0")
|
|
j = requests.get("https://api.github.com/repos/87owo/PYAS/releases/latest", headers={"Accept": "application/vnd.github+json", "User-Agent": "PYAS"}, timeout=10).json()
|
|
latest = str(j.get("tag_name") or j.get("name") or "").strip()
|
|
page = j.get("html_url") or "https://github.com/87owo/PYAS/releases"
|
|
|
|
if latest:
|
|
if not self.compare_versions(current, latest) and current != latest:
|
|
return {"has_update": True, "latest": latest, "current": current, "url": page}
|
|
return {"has_update": False, "latest": latest, "current": current, "url": page}
|
|
|
|
except Exception:
|
|
pass
|
|
return {"error": True}
|
|
|
|
####################################################################################################
|
|
|
|
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 None
|
|
|
|
####################################################################################################
|
|
|
|
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) and path.startswith(buf.value):
|
|
return path.replace(buf.value, drive, 1)
|
|
|
|
return path
|
|
|
|
def get_exe_info(self, pid):
|
|
name, file_path = None, None
|
|
h = self.kernel32.OpenProcess(0x1000, False, pid)
|
|
if h:
|
|
try:
|
|
buf = ctypes.create_unicode_buffer(1024)
|
|
size = ctypes.wintypes.DWORD(1024)
|
|
|
|
if self.kernel32.QueryFullProcessImageNameW(h, 0, buf, ctypes.byref(size)):
|
|
file_path = self.norm_path(buf.value)
|
|
if file_path:
|
|
name = os.path.basename(file_path)
|
|
|
|
elif self.psapi.GetProcessImageFileNameW(h, buf, 1024):
|
|
file_path = self.norm_path(self.device_path_to_drive(buf.value))
|
|
if file_path:
|
|
name = os.path.basename(file_path)
|
|
|
|
finally:
|
|
self.kernel32.CloseHandle(h)
|
|
|
|
return name, file_path
|
|
|
|
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_cmdline(self, h):
|
|
pbi = PROCESS_BASIC_INFORMATION()
|
|
retlen = ctypes.wintypes.ULONG(0)
|
|
if self.ntdll.NtQueryInformationProcess(h, 0, ctypes.byref(pbi), ctypes.sizeof(pbi), ctypes.byref(retlen)) != 0:
|
|
return ""
|
|
|
|
pointer_size = ctypes.sizeof(ctypes.c_void_p)
|
|
addr_pp = (int(pbi.PebBaseAddress) if pbi.PebBaseAddress else 0) + (0x20 if pointer_size == 8 else 0x10)
|
|
if not addr_pp:
|
|
return ""
|
|
|
|
read_buf = (ctypes.c_ubyte * pointer_size)()
|
|
lpBytesRead = ctypes.c_size_t(0)
|
|
if not self.kernel32.ReadProcessMemory(h, ctypes.c_void_p(addr_pp), read_buf, pointer_size, ctypes.byref(lpBytesRead)):
|
|
return ""
|
|
|
|
proc_params_address = ctypes.c_void_p.from_buffer_copy(read_buf).value
|
|
if not proc_params_address:
|
|
return ""
|
|
|
|
us = UNICODE_STRING()
|
|
if not self.kernel32.ReadProcessMemory(h, ctypes.c_void_p(int(proc_params_address) + (0x70 if pointer_size == 8 else 0x40)), ctypes.byref(us), ctypes.sizeof(us), ctypes.byref(lpBytesRead)):
|
|
return ""
|
|
if not us.Buffer or us.Length == 0:
|
|
return ""
|
|
|
|
buf = (ctypes.c_wchar * int(us.Length // 2))()
|
|
if not self.kernel32.ReadProcessMemory(h, ctypes.c_void_p(int(us.Buffer)), buf, us.Length, ctypes.byref(lpBytesRead)):
|
|
return ""
|
|
|
|
return "".join(buf)
|
|
|
|
def extract_paths_from_cmdline(self, cmdline):
|
|
if not cmdline:
|
|
return []
|
|
|
|
found = []
|
|
for m in re.finditer(r'"([^"]+)"|\'([^\']+)\'', cmdline):
|
|
path = m.group(1) or m.group(2)
|
|
if re.match(r'^[A-Za-z]:\\|^\\\\', path):
|
|
found.append(path)
|
|
|
|
if not found:
|
|
m = re.search(r'([A-Za-z]:\\[^\*?"<>\|]+\.(?:exe|dll|bat|cmd|vbs|sys|com|pif))', cmdline, re.IGNORECASE)
|
|
if m:
|
|
found.append(m.group(1))
|
|
|
|
if not found:
|
|
argc = ctypes.c_int(0)
|
|
argv = self.shell32.CommandLineToArgvW(cmdline, ctypes.byref(argc))
|
|
if argv:
|
|
args = [argv[i] for i in range(argc.value)]
|
|
self.kernel32.LocalFree(ctypes.cast(argv, ctypes.c_void_p))
|
|
patterns = [r'([A-Za-z]:\\[^"\']+)', r'(\\\\[^"\']+)', r'(\.\\[^"\']+)', r'(\./[^"\']+)', r'([A-Za-z]:/[^"\']+)', r'([^\s]*\\[^\s]+)']
|
|
|
|
for arg in args:
|
|
for p in patterns:
|
|
for match in re.finditer(p, arg): found.append(match.group(1).strip('"').strip("'"))
|
|
|
|
return list(dict.fromkeys([p.strip('"').strip("'") for p in found]))
|
|
|
|
####################################################################################################
|
|
|
|
def _enum_processes(self):
|
|
pe = PROCESSENTRY32W()
|
|
pe.dwSize = ctypes.sizeof(PROCESSENTRY32W)
|
|
snapshot = self.kernel32.CreateToolhelp32Snapshot(0x00000002, 0)
|
|
|
|
if snapshot in (-1, 0xFFFFFFFF, 0xFFFFFFFFFFFFFFFF):
|
|
return
|
|
|
|
try:
|
|
if self.kernel32.Process32FirstW(snapshot, ctypes.byref(pe)):
|
|
while True:
|
|
yield pe.th32ProcessID, pe.szExeFile
|
|
if not self.kernel32.Process32NextW(snapshot, ctypes.byref(pe)):
|
|
break
|
|
finally:
|
|
self.kernel32.CloseHandle(snapshot)
|
|
|
|
def get_process_list(self):
|
|
result = []
|
|
for pid, exe_name in self._enum_processes():
|
|
name, file_path = None, None
|
|
if pid > 4:
|
|
name, file_path = self.get_exe_info(pid)
|
|
|
|
if not name:
|
|
try:
|
|
name = exe_name
|
|
except Exception:
|
|
name = "Unknown"
|
|
|
|
result.append({"pid": pid, "name": name, "path": file_path or "None"})
|
|
|
|
return result
|
|
|
|
def get_process_list_pids(self):
|
|
return {pid for pid, _ in self._enum_processes()}
|
|
|
|
####################################################################################################
|
|
|
|
def kill_process(self, pid, expected_path=None):
|
|
try:
|
|
h = self.kernel32.OpenProcess(PYAS_PROCESS_TERMINATE_ACCESS, False, pid)
|
|
if h:
|
|
try:
|
|
file_path = self.norm_path(self.get_process_file(h))
|
|
if self.path_equal(file_path, self.file_pyas):
|
|
return False
|
|
if expected_path and not self.path_equal(file_path, expected_path):
|
|
return False
|
|
|
|
return bool(self.kernel32.TerminateProcess(h, 0))
|
|
finally:
|
|
self.kernel32.CloseHandle(h)
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "kill_process", pid=pid, detail=str(e), operate=True, success=False)
|
|
return False
|
|
|
|
####################################################################################################
|
|
|
|
def get_connections_list(self):
|
|
connections = set()
|
|
try:
|
|
size = ctypes.wintypes.DWORD()
|
|
if self.iphlpapi.GetExtendedTcpTable(None, ctypes.byref(size), True, 2, 5, 0) != 122:
|
|
return connections
|
|
|
|
buf = ctypes.create_string_buffer(size.value)
|
|
if self.iphlpapi.GetExtendedTcpTable(buf, ctypes.byref(size), True, 2, 5, 0) != 0:
|
|
return connections
|
|
|
|
num_entries = ctypes.cast(buf, ctypes.POINTER(ctypes.wintypes.DWORD)).contents.value
|
|
for i in range(num_entries):
|
|
row = MIB_TCPROW_OWNER_PID.from_address(ctypes.addressof(buf) + ctypes.sizeof(ctypes.wintypes.DWORD) + i * ctypes.sizeof(MIB_TCPROW_OWNER_PID))
|
|
connections.add((row.dwOwningPid, row.dwRemoteAddr, row.dwRemotePort))
|
|
|
|
except Exception:
|
|
pass
|
|
return connections
|
|
|
|
def get_traffic_list(self):
|
|
conns = self.get_connections_list()
|
|
conn_map = {}
|
|
for pid, _, _ in conns:
|
|
conn_map[pid] = conn_map.get(pid, 0) + 1
|
|
|
|
current_io = {}
|
|
current_time = time.time()
|
|
|
|
with self.lock_io:
|
|
time_diff = current_time - self.last_io_time
|
|
if time_diff <= 0:
|
|
time_diff = 1
|
|
|
|
self.last_io_time = current_time
|
|
old_counters = self.last_io_counters.copy()
|
|
|
|
result = []
|
|
exist_process = self.get_process_list_pids()
|
|
|
|
for pid in exist_process:
|
|
name, file_path = "Unknown", ""
|
|
if pid > 4:
|
|
name, file_path = self.get_exe_info(pid)
|
|
else:
|
|
name = "System"
|
|
|
|
down_speed, up_speed = 0, 0
|
|
h = self.kernel32.OpenProcess(0x1000, False, pid)
|
|
if h:
|
|
try:
|
|
io = IO_COUNTERS()
|
|
if self.kernel32.GetProcessIoCounters(h, ctypes.byref(io)):
|
|
current_io[pid] = (io.ReadTransferCount, io.WriteTransferCount)
|
|
|
|
if pid in old_counters:
|
|
old_read, old_write = old_counters[pid]
|
|
down_speed = max(0, (io.ReadTransferCount - old_read) / time_diff)
|
|
up_speed = max(0, (io.WriteTransferCount - old_write) / time_diff)
|
|
|
|
finally:
|
|
self.kernel32.CloseHandle(h)
|
|
|
|
count = conn_map.get(pid, 0)
|
|
if count > 0 or down_speed > 0 or up_speed > 0:
|
|
result.append({"pid": pid, "name": name, "path": file_path or "None", "down": int(down_speed), "up": int(up_speed), "conn": count})
|
|
|
|
with self.lock_io:
|
|
self.last_io_counters = current_io
|
|
return result
|
|
|
|
####################################################################################################
|
|
|
|
def _traverse_delete(self, path):
|
|
deleted = 0
|
|
if not path or not os.path.exists(path):
|
|
return deleted
|
|
|
|
try:
|
|
items = os.listdir(path)
|
|
except Exception:
|
|
return deleted
|
|
|
|
for fd in items:
|
|
file = os.path.join(path, fd)
|
|
try:
|
|
if os.path.isdir(file):
|
|
if self._is_reparse_point(file):
|
|
continue
|
|
|
|
deleted += self._traverse_delete(file)
|
|
try:
|
|
os.rmdir(file)
|
|
except Exception:
|
|
pass
|
|
|
|
else:
|
|
if file.lower().endswith(('.sys', '.dll', '.exe', '.ini', '.dat')):
|
|
continue
|
|
|
|
size = os.path.getsize(file)
|
|
os.remove(file)
|
|
deleted += size
|
|
|
|
except Exception:
|
|
continue
|
|
|
|
return deleted
|
|
|
|
def _get_junk_dirs(self):
|
|
return [
|
|
self.path_temp,
|
|
self.path_systemp,
|
|
os.path.join(self.path_system, "SoftwareDistribution", "Download")
|
|
]
|
|
|
|
def _yield_log_files(self):
|
|
log_dir = os.path.join(self.path_system, "Logs")
|
|
if os.path.exists(log_dir):
|
|
for root, _, files in os.walk(log_dir):
|
|
for file in files:
|
|
if file.lower().endswith(('.log', '.etl', '.evtx')):
|
|
yield os.path.join(root, file)
|
|
|
|
def _get_recycle_bin_size(self):
|
|
try:
|
|
info = SHQUERYRBINFO()
|
|
info.cbSize = ctypes.sizeof(SHQUERYRBINFO)
|
|
|
|
if self.shell32.SHQueryRecycleBinW(None, ctypes.byref(info)) == 0 and info.i64Size > 0:
|
|
return info.i64Size
|
|
except Exception:
|
|
pass
|
|
|
|
return 0
|
|
|
|
def scan_system_junk(self):
|
|
junk_list = []
|
|
try:
|
|
for path in self._get_junk_dirs():
|
|
if os.path.exists(path):
|
|
for root, _, files in os.walk(path):
|
|
for file in files:
|
|
if not file.lower().endswith(('.sys', '.dll', '.exe', '.ini', '.dat')):
|
|
try:
|
|
fp = os.path.join(root, file)
|
|
junk_list.append({"path": fp, "size": os.path.getsize(fp)})
|
|
except Exception:
|
|
pass
|
|
|
|
for fp in self._yield_log_files():
|
|
try:
|
|
junk_list.append({"path": fp, "size": os.path.getsize(fp)})
|
|
except Exception:
|
|
pass
|
|
|
|
rb_size = self._get_recycle_bin_size()
|
|
if rb_size > 0:
|
|
junk_list.append({"path": "Recycle Bin", "size": rb_size})
|
|
|
|
return junk_list
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "scan_system_junk", detail=str(e), success=False)
|
|
return []
|
|
|
|
def clean_system_junk(self, paths_to_delete=None):
|
|
total_deleted = 0
|
|
try:
|
|
if paths_to_delete is not None:
|
|
for path in paths_to_delete:
|
|
if path == "Recycle Bin":
|
|
rb_size = self._get_recycle_bin_size()
|
|
if rb_size > 0 and self.shell32.SHEmptyRecycleBinW(None, None, 7) == 0:
|
|
total_deleted += rb_size
|
|
continue
|
|
|
|
if path.lower().endswith(('.sys', '.dll', '.exe', '.ini', '.dat')):
|
|
continue
|
|
|
|
try:
|
|
size = os.path.getsize(path)
|
|
os.remove(path)
|
|
total_deleted += size
|
|
except Exception:
|
|
pass
|
|
else:
|
|
for path in self._get_junk_dirs():
|
|
total_deleted += self._traverse_delete(path)
|
|
|
|
for fp in self._yield_log_files():
|
|
try:
|
|
size = os.path.getsize(fp)
|
|
os.remove(fp)
|
|
total_deleted += size
|
|
except Exception:
|
|
pass
|
|
|
|
rb_size = self._get_recycle_bin_size()
|
|
if rb_size > 0 and self.shell32.SHEmptyRecycleBinW(None, None, 7) == 0:
|
|
total_deleted += rb_size
|
|
|
|
try:
|
|
for log_type in ["Application", "Security", "Setup", "System"]:
|
|
log_path = os.path.join(self.path_system, "System32", "winevt", "Logs", f"{log_type}.evtx")
|
|
|
|
size = os.path.getsize(log_path) if os.path.exists(log_path) else 0
|
|
if self._clear_event_log(log_type):
|
|
total_deleted += size
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
self.write_log("INFO", "Clean Junk", detail=f"Deleted {total_deleted // 1024} KB", operate=True)
|
|
return total_deleted
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "clean_system_junk", detail=str(e), operate=True, success=False)
|
|
return 0
|
|
|
|
####################################################################################################
|
|
|
|
def optimize_memory(self, icon=None, item=None):
|
|
def _optimize():
|
|
try:
|
|
hwnd = self.user32.GetForegroundWindow()
|
|
fg_pid = ctypes.wintypes.DWORD(0)
|
|
if hwnd:
|
|
self.user32.GetWindowThreadProcessId(hwnd, ctypes.byref(fg_pid))
|
|
|
|
pids = self.get_process_list_pids()
|
|
optimized_count = 0
|
|
|
|
skip_set = {
|
|
"dwm.exe", "explorer.exe", "csrss.exe", "smss.exe", "winlogon.exe",
|
|
"lsass.exe", "services.exe", "svchost.exe", "wininit.exe", "audiodg.exe",
|
|
"spoolsv.exe", "sihost.exe", "fontdrvhost.exe", "taskmgr.exe"
|
|
}
|
|
|
|
for pid in pids:
|
|
if pid <= 4 or pid == self.pid_pyas or pid == fg_pid.value:
|
|
continue
|
|
|
|
name, _ = self.get_exe_info(pid)
|
|
if name and name.lower() in skip_set:
|
|
continue
|
|
|
|
h = self.kernel32.OpenProcess(0x0100, False, pid)
|
|
if h:
|
|
try:
|
|
if self.kernel32.SetProcessWorkingSetSize(h, ctypes.c_size_t(-1), ctypes.c_size_t(-1)):
|
|
optimized_count += 1
|
|
finally:
|
|
self.kernel32.CloseHandle(h)
|
|
|
|
title = self._loc({
|
|
"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"
|
|
})
|
|
|
|
msg = self._loc({
|
|
"traditional_switch": f"已釋放 {optimized_count} 個背景進程的記憶體",
|
|
"simplified_switch": f"已释放 {optimized_count} 个后台进程的内存",
|
|
"english_switch": f"Freed memory for {optimized_count} background processes",
|
|
"japanese_switch": f"{optimized_count} 個のバックグラウンドプロセスのメモリを解放しました",
|
|
"korean_switch": f"{optimized_count}개 백그라운드 프로세스 메모리 확보",
|
|
"french_switch": f"Mémoire libérée pour {optimized_count} processus",
|
|
"spanish_switch": f"Memoria liberada para {optimized_count} procesos",
|
|
"hindi_switch": f"{optimized_count} पृष्ठभूमि प्रक्रियाओं के लिए मेमोरी मुक्त की गई",
|
|
"arabic_switch": f"تم تحرير الذاكرة لـ {optimized_count} من العمليات في الخلفية",
|
|
"russian_switch": f"Освобождена память {optimized_count} фоновых процессов",
|
|
"slovenian_switch": f"Sproščen pomnilnik za {optimized_count} procesov v ozadju"
|
|
})
|
|
|
|
self.show_notification(title, msg)
|
|
self.write_log("INFO", "Memory Boost", detail=f"Optimized {optimized_count} processes", operate=True)
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "optimize_memory", detail=str(e), success=False)
|
|
|
|
threading.Thread(target=_optimize, daemon=True).start()
|
|
|
|
####################################################################################################
|
|
|
|
def get_startup_list(self):
|
|
items = []
|
|
|
|
def get_reg_startup():
|
|
res = []
|
|
paths = [
|
|
(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Run", "enabled"),
|
|
(winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Run", "enabled"),
|
|
(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Run_Disabled", "disabled"),
|
|
(winreg.HKEY_CURRENT_USER, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Run_Disabled", "disabled")
|
|
]
|
|
|
|
for root, path, status in paths:
|
|
try:
|
|
with winreg.OpenKey(root, path, 0, winreg.KEY_READ) as reg:
|
|
i = 0
|
|
while True:
|
|
try:
|
|
name, val, _ = winreg.EnumValue(reg, i)
|
|
file_path = self.extract_paths_from_cmdline(val)
|
|
fp = file_path[0] if file_path else val
|
|
res.append({"id": f"reg|{root}|{path}|{name}", "type": "type_reg", "name": name, "status": status, "path": fp})
|
|
i += 1
|
|
|
|
except OSError:
|
|
break
|
|
except Exception:
|
|
pass
|
|
return res
|
|
|
|
def get_services():
|
|
res = []
|
|
try:
|
|
with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SYSTEM\CurrentControlSet\Services", 0, winreg.KEY_READ) as reg:
|
|
i = 0
|
|
while True:
|
|
try:
|
|
name = winreg.EnumKey(reg, i)
|
|
i += 1
|
|
try:
|
|
with winreg.OpenKey(reg, name, 0, winreg.KEY_READ) as sub:
|
|
stype, _ = winreg.QueryValueEx(sub, "Type")
|
|
if stype not in (16, 32, 256):
|
|
continue
|
|
|
|
start, _ = winreg.QueryValueEx(sub, "Start")
|
|
if start not in (2, 3, 4):
|
|
continue
|
|
|
|
path, _ = winreg.QueryValueEx(sub, "ImagePath")
|
|
if path and "svchost" not in path.lower() and "windows" not in path.lower():
|
|
fp = self.extract_paths_from_cmdline(path)
|
|
fpp = fp[0] if fp else path
|
|
status = "enabled" if start == 2 else "disabled"
|
|
res.append({"id": f"srv|{name}", "type": "type_srv", "name": name, "status": status, "path": fpp})
|
|
|
|
except Exception:
|
|
pass
|
|
except OSError:
|
|
break
|
|
|
|
except Exception:
|
|
pass
|
|
return res
|
|
|
|
def get_tasks():
|
|
res = []
|
|
try:
|
|
proc = self._run_windows_tool(
|
|
("schtasks.exe",),
|
|
["/query", "/fo", "csv", "/v"],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
if proc is not None and proc.returncode == 0:
|
|
reader = csv.reader(io.StringIO(proc.stdout))
|
|
next(reader, None)
|
|
|
|
for row in reader:
|
|
if len(row) > 8 and row[1].strip() and row[8].strip() and row[8] != "N/A":
|
|
name = row[1].split('\\')[-1]
|
|
path = row[8]
|
|
status = row[3]
|
|
|
|
if path and "windows" not in path.lower():
|
|
fp = self.extract_paths_from_cmdline(path)
|
|
fpp = fp[0] if fp else path
|
|
res.append({"id": f"tsk|{row[1]}", "type": "type_tsk", "name": name, "status": "disabled" if status.lower() == "disabled" else "enabled", "path": fpp})
|
|
return res
|
|
|
|
command = (
|
|
"[Console]::OutputEncoding = [Text.UTF8Encoding]::new(); "
|
|
"Get-ScheduledTask -ErrorAction SilentlyContinue | ForEach-Object { "
|
|
"$Action = @($_.Actions)[0]; "
|
|
"[pscustomobject]@{Name=$_.TaskName; FullName=($_.TaskPath + $_.TaskName); "
|
|
"State=[string]$_.State; Execute=$Action.Execute; Arguments=$Action.Arguments} "
|
|
"} | ConvertTo-Json -Compress"
|
|
)
|
|
proc = self._run_powershell(command, capture_output=True, text=True, encoding="utf-8", errors="replace")
|
|
if not proc or proc.returncode != 0 or not proc.stdout.strip():
|
|
return res
|
|
|
|
task_data = json.loads(proc.stdout)
|
|
if isinstance(task_data, dict):
|
|
task_data = [task_data]
|
|
|
|
for task in task_data:
|
|
name = str(task.get("Name") or "").strip()
|
|
full_name = str(task.get("FullName") or name).strip()
|
|
execute = str(task.get("Execute") or "").strip()
|
|
arguments = str(task.get("Arguments") or "").strip()
|
|
command_line = " ".join(part for part in (execute, arguments) if part)
|
|
if not name or not command_line or "windows" in command_line.lower():
|
|
continue
|
|
|
|
paths = self.extract_paths_from_cmdline(command_line)
|
|
path = paths[0] if paths else execute or command_line
|
|
status = "disabled" if str(task.get("State") or "").lower() == "disabled" else "enabled"
|
|
res.append({"id": f"tsk|{full_name}", "type": "type_tsk", "name": name, "status": status, "path": path})
|
|
|
|
except Exception:
|
|
pass
|
|
return res
|
|
|
|
with ThreadPoolExecutor(max_workers=3) as executor:
|
|
f1 = executor.submit(get_reg_startup)
|
|
f2 = executor.submit(get_services)
|
|
f3 = executor.submit(get_tasks)
|
|
|
|
items.extend(f1.result())
|
|
items.extend(f2.result())
|
|
items.extend(f3.result())
|
|
|
|
return items
|
|
|
|
def manage_startup(self, items, action):
|
|
def process_item(item_id):
|
|
try:
|
|
parts = item_id.split('|', 3)
|
|
stype = parts[0]
|
|
|
|
if stype == "reg":
|
|
root, old_path, name = int(parts[1]), parts[2], parts[3]
|
|
if action == "delete":
|
|
return self._reg_delete(root, old_path, name)
|
|
|
|
else:
|
|
new_path = old_path.replace("_Disabled", "") if action == "enable" else (old_path if "_Disabled" in old_path else old_path + "_Disabled")
|
|
if old_path != new_path:
|
|
val = self._reg_read(root, old_path, name)
|
|
if val:
|
|
return self._reg_write(root, new_path, name, winreg.REG_SZ, val) and self._reg_delete(root, old_path, name)
|
|
return True
|
|
|
|
elif stype == "srv":
|
|
name = parts[1]
|
|
if action == "delete":
|
|
result = self._run_windows_tool(("sc.exe",), ["delete", name], capture_output=True)
|
|
|
|
else:
|
|
mode = "auto" if action == "enable" else "disabled"
|
|
result = self._run_windows_tool(("sc.exe",), ["config", name, "start=", mode], capture_output=True)
|
|
|
|
if result is not None and result.returncode == 0:
|
|
return True
|
|
return self._manage_service_native(name, action)
|
|
|
|
elif stype == "tsk":
|
|
name = item_id.partition('|')[2]
|
|
return self._manage_scheduled_task(name, action)
|
|
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.write_log("WARN", "manage_startup", detail=str(e), success=False)
|
|
return False
|
|
|
|
with ThreadPoolExecutor(max_workers=5) as executor:
|
|
results = list(executor.map(process_item, items))
|
|
|
|
success = all(results) if results else True
|
|
if not success:
|
|
self.write_log("WARN", "manage_startup", detail="One or more startup items could not be updated", success=False)
|
|
return success
|