764 lines
28 KiB
Python
764 lines
28 KiB
Python
from PYAS_Diagnostics import log_exception
|
|
import time
|
|
import copy
|
|
import ctypes
|
|
import ctypes.wintypes
|
|
from PYAS_WinAPI import POINT, RECT
|
|
import os
|
|
import uuid
|
|
import math
|
|
|
|
|
|
def normalized_path(path):
|
|
return os.path.normcase(os.path.normpath(path)) if isinstance(path, str) and path else ""
|
|
|
|
|
|
def signature_matches(expected, actual):
|
|
if not isinstance(expected, dict) or not isinstance(actual, dict):
|
|
return False
|
|
|
|
for key in ("class", "style", "exstyle", "children", "owner"):
|
|
if key not in expected or expected[key] != actual.get(key):
|
|
return False
|
|
|
|
for key in ("width", "height"):
|
|
size = expected.get(key)
|
|
current = actual.get(key)
|
|
|
|
if not isinstance(size, (int, float)) or not isinstance(current, (int, float)):
|
|
return False
|
|
|
|
if (
|
|
not math.isfinite(size)
|
|
or not math.isfinite(current)
|
|
or size <= 0
|
|
or current <= 0
|
|
or abs(size - current) > max(8, size * 0.08)
|
|
):
|
|
return False
|
|
|
|
return True
|
|
|
|
|
|
def is_main_anchor(window, target):
|
|
return (
|
|
window.get("hwnd") != target.get("hwnd")
|
|
and window.get("pid") == target.get("pid")
|
|
and normalized_path(window.get("path")) == normalized_path(target.get("path"))
|
|
and not window.get("fingerprint", {}).get("owner")
|
|
and not window.get("fingerprint", {}).get("exstyle", 0) & 0x80
|
|
and window.get("fingerprint", {}).get("style", 0) & 0x00030000 == 0x00030000
|
|
and window.get("fingerprint", {}).get("width", 0)
|
|
* window.get("fingerprint", {}).get("height", 0)
|
|
> target.get("fingerprint", {}).get("width", 0)
|
|
* target.get("fingerprint", {}).get("height", 0)
|
|
* 1.25
|
|
)
|
|
|
|
|
|
def secondary_candidate(target):
|
|
fingerprint = target.get("fingerprint", {})
|
|
return bool(
|
|
fingerprint.get("owner")
|
|
or fingerprint.get("exstyle", 0) & 0x80
|
|
or not fingerprint.get("style", 0) & 0x00030000
|
|
)
|
|
|
|
|
|
def build_popup_rule(target, windows):
|
|
if not target or not target.get("path") or not target.get("fingerprint", {}).get("class"):
|
|
return None
|
|
|
|
fingerprint = target["fingerprint"]
|
|
peers = [
|
|
window
|
|
for window in windows
|
|
if window.get("pid") == target.get("pid")
|
|
and normalized_path(window.get("path")) == normalized_path(target.get("path"))
|
|
and window["hwnd"] != target["hwnd"]
|
|
]
|
|
identical = any(
|
|
all(
|
|
fingerprint.get(key) == window.get("fingerprint", {}).get(key)
|
|
for key in ("class", "style", "exstyle", "children", "owner")
|
|
)
|
|
for window in peers
|
|
)
|
|
anchors = [window for window in peers if is_main_anchor(window, target)]
|
|
distinctive = secondary_candidate(target) or bool(anchors)
|
|
mode = (
|
|
"adaptive"
|
|
if isinstance(fingerprint.get("children"), list)
|
|
and signature_matches(fingerprint, fingerprint)
|
|
and distinctive
|
|
and not identical
|
|
and not target.get("modal")
|
|
else "title"
|
|
)
|
|
return {
|
|
"id": uuid.uuid4().hex,
|
|
"schema_version": 2,
|
|
"exe": target["exe"],
|
|
"path": target["path"],
|
|
"class": fingerprint["class"],
|
|
"title": target.get("title", ""),
|
|
"fingerprint": fingerprint,
|
|
"match_mode": mode,
|
|
"protected": [
|
|
window["fingerprint"]
|
|
for window in peers
|
|
if not signature_matches(fingerprint, window.get("fingerprint"))
|
|
],
|
|
}
|
|
|
|
|
|
def popup_rule_matches(rule, target, windows):
|
|
if not isinstance(rule, dict) or not target:
|
|
return False
|
|
|
|
if rule.get("schema_version") != 2:
|
|
return bool(
|
|
rule.get("exe") == target.get("exe")
|
|
and rule.get("class") == target.get("fingerprint", {}).get("class")
|
|
and rule.get("title")
|
|
and rule["title"] == target.get("title")
|
|
and not target.get("modal")
|
|
)
|
|
|
|
if (
|
|
not rule.get("id")
|
|
or not normalized_path(rule.get("path"))
|
|
or normalized_path(rule["path"]) != normalized_path(target.get("path"))
|
|
):
|
|
return False
|
|
|
|
fingerprint = target.get("fingerprint", {})
|
|
|
|
if not rule.get("class") or rule["class"] != fingerprint.get("class"):
|
|
return False
|
|
|
|
if any(signature_matches(protected, fingerprint) for protected in rule.get("protected", [])):
|
|
return False
|
|
|
|
expected = rule.get("fingerprint", {})
|
|
structural_match = signature_matches(expected, fingerprint)
|
|
title_match = bool(rule.get("title") and rule["title"] == target.get("title"))
|
|
|
|
if rule.get("match_mode", "adaptive") == "title":
|
|
if not title_match or not all(
|
|
expected.get(key) == fingerprint.get(key) for key in ("style", "exstyle", "owner")
|
|
):
|
|
return False
|
|
|
|
return not any(
|
|
window.get("hwnd") != target.get("hwnd")
|
|
and window.get("pid") == target.get("pid")
|
|
and normalized_path(window.get("path")) == normalized_path(target.get("path"))
|
|
and window.get("title") == target.get("title")
|
|
and window.get("fingerprint", {}).get("class") == rule["class"]
|
|
and all(
|
|
window.get("fingerprint", {}).get(key) == fingerprint.get(key)
|
|
for key in ("style", "exstyle", "owner")
|
|
)
|
|
for window in windows
|
|
)
|
|
|
|
if not structural_match or target.get("modal"):
|
|
return False
|
|
|
|
peers = [
|
|
window
|
|
for window in windows
|
|
if window.get("pid") == target.get("pid")
|
|
and normalized_path(window.get("path")) == normalized_path(target.get("path"))
|
|
and window["hwnd"] != target["hwnd"]
|
|
]
|
|
identical = [
|
|
window for window in peers if signature_matches(expected, window.get("fingerprint"))
|
|
]
|
|
|
|
if identical:
|
|
return title_match and not any(
|
|
window.get("title") == target.get("title") for window in identical
|
|
)
|
|
|
|
return True
|
|
|
|
|
|
class PopupWindowMixin:
|
|
def capture_popup_window(self):
|
|
self._init_popup_api()
|
|
overlay = self.user32.CreateWindowExW(
|
|
0x000800A8, "STATIC", "", 0x98000004, 0, 0, 0, 0, None, None, None, None
|
|
)
|
|
|
|
if overlay:
|
|
self.user32.SetLayeredWindowAttributes(overlay, 0, 120, 2)
|
|
|
|
target_hwnd = None
|
|
last_hwnd = None
|
|
msg = ctypes.wintypes.MSG()
|
|
|
|
start_time = time.monotonic()
|
|
|
|
try:
|
|
while self.user32.GetAsyncKeyState(0x01) & 0x8000:
|
|
if time.monotonic() - start_time >= 30:
|
|
return None
|
|
|
|
time.sleep(0.01)
|
|
|
|
while time.monotonic() - start_time < 30:
|
|
if self.user32.GetAsyncKeyState(0x1B) & 0x8000:
|
|
break
|
|
|
|
pt = POINT()
|
|
self.user32.GetCursorPos(ctypes.byref(pt))
|
|
|
|
hwnd = self.user32.WindowFromPoint(pt)
|
|
root_hwnd = self.user32.GetAncestor(hwnd, 2)
|
|
|
|
if not root_hwnd:
|
|
root_hwnd = hwnd
|
|
|
|
if root_hwnd and root_hwnd != overlay:
|
|
if root_hwnd != last_hwnd:
|
|
last_hwnd = root_hwnd
|
|
rect = RECT()
|
|
self.user32.GetWindowRect(root_hwnd, ctypes.byref(rect))
|
|
self.user32.SetWindowPos(
|
|
overlay,
|
|
-1,
|
|
rect.left,
|
|
rect.top,
|
|
rect.right - rect.left,
|
|
rect.bottom - rect.top,
|
|
0x0050,
|
|
)
|
|
|
|
if self.user32.GetAsyncKeyState(0x01) & 0x8000:
|
|
target_hwnd = root_hwnd
|
|
break
|
|
|
|
while self.user32.PeekMessageW(ctypes.byref(msg), 0, 0, 0, 1):
|
|
self.user32.TranslateMessage(ctypes.byref(msg))
|
|
self.user32.DispatchMessageW(ctypes.byref(msg))
|
|
|
|
time.sleep(0.02)
|
|
finally:
|
|
if overlay:
|
|
self.user32.DestroyWindow(overlay)
|
|
|
|
if not target_hwnd:
|
|
return None
|
|
|
|
pid = ctypes.c_ulong(0)
|
|
self.user32.GetWindowThreadProcessId(target_hwnd, ctypes.byref(pid))
|
|
|
|
if pid.value != self.pid_pyas and pid.value > 4:
|
|
length = self.user32.GetWindowTextLengthW(target_hwnd)
|
|
title = ctypes.create_unicode_buffer(length + 1)
|
|
class_name = ctypes.create_unicode_buffer(256)
|
|
|
|
self.user32.GetWindowTextW(target_hwnd, title, length + 1)
|
|
self.user32.GetClassNameW(target_hwnd, class_name, 256)
|
|
|
|
proc_name, _ = self.get_exe_info(pid.value)
|
|
t_str, c_str = str(title.value), str(class_name.value)
|
|
|
|
if proc_name and not any(
|
|
item.get("exe") == proc_name or item.get("class") == c_str
|
|
for item in self.pass_windows
|
|
):
|
|
target = self._popup_window_snapshot(target_hwnd)
|
|
windows = self._popup_windows({normalized_path(target["path"])}) if target else []
|
|
rule = build_popup_rule(target, windows)
|
|
|
|
if not rule:
|
|
return {"error": "popup_capture_failed"}
|
|
|
|
token = (time.monotonic_ns() & 0x7FFFFFFF) or 1
|
|
selected = {"hwnd": target_hwnd, "pid": target["pid"], "token": token}
|
|
|
|
if not self.user32.SetPropW(target_hwnd, "PYAS_Popup_Selected", token):
|
|
return {"error": "popup_capture_failed"}
|
|
|
|
with self.lock_config:
|
|
previous = getattr(self, "_popup_selected", {})
|
|
active = {
|
|
item.get("id")
|
|
for item in self.pyas_config.get("block_list", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
|
|
for identifier, record in list(previous.items()):
|
|
if identifier not in active:
|
|
if (
|
|
self.user32.GetPropW(record["hwnd"], "PYAS_Popup_Selected")
|
|
== record["token"]
|
|
):
|
|
self.user32.RemovePropW(record["hwnd"], "PYAS_Popup_Selected")
|
|
|
|
previous.pop(identifier, None)
|
|
|
|
previous[rule["id"]] = selected
|
|
self._popup_selected = previous
|
|
self._captured_popup_rules = {rule["id"]: copy.deepcopy(rule)}
|
|
|
|
return rule
|
|
|
|
return {"error": "popup_capture_failed"}
|
|
|
|
def _init_popup_api(self):
|
|
if getattr(self, "_popup_api_ready", False):
|
|
return
|
|
|
|
self._popup_enum_type = ctypes.WINFUNCTYPE(
|
|
ctypes.wintypes.BOOL, ctypes.wintypes.HWND, ctypes.wintypes.LPARAM
|
|
)
|
|
self.user32.CreateWindowExW.argtypes = [
|
|
ctypes.wintypes.DWORD,
|
|
ctypes.c_wchar_p,
|
|
ctypes.c_wchar_p,
|
|
ctypes.wintypes.DWORD,
|
|
ctypes.c_int,
|
|
ctypes.c_int,
|
|
ctypes.c_int,
|
|
ctypes.c_int,
|
|
ctypes.wintypes.HWND,
|
|
ctypes.wintypes.HMENU,
|
|
ctypes.wintypes.HINSTANCE,
|
|
ctypes.c_void_p,
|
|
]
|
|
self.user32.CreateWindowExW.restype = ctypes.wintypes.HWND
|
|
self.user32.DestroyWindow.argtypes = [ctypes.wintypes.HWND]
|
|
self.user32.DestroyWindow.restype = ctypes.wintypes.BOOL
|
|
self.user32.PeekMessageW.argtypes = [
|
|
ctypes.POINTER(ctypes.wintypes.MSG),
|
|
ctypes.wintypes.HWND,
|
|
ctypes.c_uint,
|
|
ctypes.c_uint,
|
|
ctypes.c_uint,
|
|
]
|
|
self.user32.PeekMessageW.restype = ctypes.wintypes.BOOL
|
|
self.user32.TranslateMessage.argtypes = [ctypes.POINTER(ctypes.wintypes.MSG)]
|
|
self.user32.TranslateMessage.restype = ctypes.wintypes.BOOL
|
|
self.user32.DispatchMessageW.argtypes = [ctypes.POINTER(ctypes.wintypes.MSG)]
|
|
self.user32.DispatchMessageW.restype = ctypes.wintypes.LPARAM
|
|
|
|
for name in ("EnumWindows", "EnumChildWindows"):
|
|
function = getattr(self.user32, name)
|
|
function.argtypes = (
|
|
[self._popup_enum_type, ctypes.wintypes.LPARAM]
|
|
if name == "EnumWindows"
|
|
else [ctypes.wintypes.HWND, self._popup_enum_type, ctypes.wintypes.LPARAM]
|
|
)
|
|
function.restype = ctypes.wintypes.BOOL
|
|
|
|
self._popup_get_style = getattr(
|
|
self.user32, "GetWindowLongPtrW", self.user32.GetWindowLongW
|
|
)
|
|
self._popup_get_style.argtypes = [ctypes.wintypes.HWND, ctypes.c_int]
|
|
self._popup_get_style.restype = ctypes.c_ssize_t
|
|
|
|
for name in ("IsWindow", "IsWindowVisible", "IsWindowEnabled", "IsIconic"):
|
|
function = getattr(self.user32, name)
|
|
function.argtypes = [ctypes.wintypes.HWND]
|
|
function.restype = ctypes.wintypes.BOOL
|
|
|
|
self.user32.GetWindow.argtypes = [ctypes.wintypes.HWND, ctypes.c_uint]
|
|
self.user32.GetWindow.restype = ctypes.wintypes.HWND
|
|
self.user32.GetClientRect.argtypes = [ctypes.wintypes.HWND, ctypes.POINTER(RECT)]
|
|
self.user32.GetClientRect.restype = ctypes.wintypes.BOOL
|
|
|
|
class WINDOWPLACEMENT(ctypes.Structure):
|
|
_fields_ = [
|
|
("length", ctypes.wintypes.DWORD),
|
|
("flags", ctypes.wintypes.DWORD),
|
|
("showCmd", ctypes.wintypes.DWORD),
|
|
("ptMinPosition", POINT),
|
|
("ptMaxPosition", POINT),
|
|
("rcNormalPosition", RECT),
|
|
]
|
|
|
|
self._popup_placement_type = WINDOWPLACEMENT
|
|
self.user32.GetWindowPlacement.argtypes = [
|
|
ctypes.wintypes.HWND,
|
|
ctypes.POINTER(WINDOWPLACEMENT),
|
|
]
|
|
self.user32.GetWindowPlacement.restype = ctypes.wintypes.BOOL
|
|
self.user32.GetWindowThreadProcessId.argtypes = [
|
|
ctypes.wintypes.HWND,
|
|
ctypes.POINTER(ctypes.wintypes.DWORD),
|
|
]
|
|
self.user32.GetWindowThreadProcessId.restype = ctypes.wintypes.DWORD
|
|
self.user32.GetWindowTextLengthW.argtypes = [ctypes.wintypes.HWND]
|
|
self.user32.GetWindowTextLengthW.restype = ctypes.c_int
|
|
self.user32.GetWindowTextW.argtypes = [ctypes.wintypes.HWND, ctypes.c_wchar_p, ctypes.c_int]
|
|
self.user32.GetClassNameW.argtypes = [ctypes.wintypes.HWND, ctypes.c_wchar_p, ctypes.c_int]
|
|
self.user32.ShowWindowAsync.argtypes = [ctypes.wintypes.HWND, ctypes.c_int]
|
|
self.user32.ShowWindowAsync.restype = ctypes.wintypes.BOOL
|
|
self.user32.SetPropW.argtypes = [
|
|
ctypes.wintypes.HWND,
|
|
ctypes.c_wchar_p,
|
|
ctypes.wintypes.HANDLE,
|
|
]
|
|
self.user32.SetPropW.restype = ctypes.wintypes.BOOL
|
|
self.user32.GetPropW.argtypes = [ctypes.wintypes.HWND, ctypes.c_wchar_p]
|
|
self.user32.GetPropW.restype = ctypes.wintypes.HANDLE
|
|
self.user32.RemovePropW.argtypes = [ctypes.wintypes.HWND, ctypes.c_wchar_p]
|
|
self.user32.RemovePropW.restype = ctypes.wintypes.HANDLE
|
|
|
|
if hasattr(self.user32, "GetDpiForWindow"):
|
|
self.user32.GetDpiForWindow.argtypes = [ctypes.wintypes.HWND]
|
|
self.user32.GetDpiForWindow.restype = ctypes.c_uint
|
|
|
|
self._popup_api_ready = True
|
|
|
|
def _popup_window_snapshot(self, hwnd, process_info=None):
|
|
if not self.user32.IsWindow(hwnd):
|
|
return None
|
|
|
|
pid = ctypes.wintypes.DWORD()
|
|
self.user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
|
|
|
if pid.value <= 4 or pid.value == self.pid_pyas:
|
|
return None
|
|
|
|
process_info = {} if process_info is None else process_info
|
|
|
|
if pid.value not in process_info:
|
|
process_info[pid.value] = self.get_exe_info(pid.value)
|
|
|
|
exe, path = process_info[pid.value]
|
|
|
|
if not exe or not path:
|
|
return None
|
|
|
|
class_name = ctypes.create_unicode_buffer(256)
|
|
|
|
if not self.user32.GetClassNameW(hwnd, class_name, 256):
|
|
return None
|
|
|
|
if any(
|
|
item.get("exe") == exe or item.get("class") == class_name.value
|
|
for item in self.pass_windows
|
|
):
|
|
return None
|
|
|
|
rect = RECT()
|
|
|
|
if not self.user32.GetClientRect(hwnd, ctypes.byref(rect)):
|
|
self.user32.GetWindowRect(hwnd, ctypes.byref(rect))
|
|
|
|
if self.user32.IsIconic(hwnd):
|
|
placement = self._popup_placement_type()
|
|
placement.length = ctypes.sizeof(placement)
|
|
|
|
if self.user32.GetWindowPlacement(hwnd, ctypes.byref(placement)):
|
|
rect = placement.rcNormalPosition
|
|
|
|
dpi = self.user32.GetDpiForWindow(hwnd) if hasattr(self.user32, "GetDpiForWindow") else 96
|
|
dpi = dpi or 96
|
|
classes = {}
|
|
child_count = 0
|
|
|
|
def enum_child(child, parameter):
|
|
nonlocal child_count
|
|
child_count += 1
|
|
|
|
if child_count > 128:
|
|
return False
|
|
|
|
name = ctypes.create_unicode_buffer(256)
|
|
|
|
if not self.user32.GetClassNameW(child, name, 256):
|
|
child_count = 129
|
|
return False
|
|
|
|
classes[name.value] = classes.get(name.value, 0) + 1
|
|
return True
|
|
|
|
callback = self._popup_enum_type(enum_child)
|
|
self.user32.EnumChildWindows(hwnd, callback, 0)
|
|
children = (
|
|
None
|
|
if child_count > 128
|
|
else [[name, count] for name, count in sorted(classes.items())]
|
|
)
|
|
owner_hwnd = self.user32.GetWindow(hwnd, 4)
|
|
owner = None
|
|
|
|
if owner_hwnd:
|
|
owner_pid = ctypes.wintypes.DWORD()
|
|
self.user32.GetWindowThreadProcessId(owner_hwnd, ctypes.byref(owner_pid))
|
|
|
|
if owner_pid.value not in process_info:
|
|
process_info[owner_pid.value] = self.get_exe_info(owner_pid.value)
|
|
|
|
_, owner_path = process_info[owner_pid.value]
|
|
owner_class = ctypes.create_unicode_buffer(256)
|
|
self.user32.GetClassNameW(owner_hwnd, owner_class, 256)
|
|
owner = {"path": normalized_path(owner_path), "class": owner_class.value}
|
|
|
|
title = ctypes.create_unicode_buffer(min(self.user32.GetWindowTextLengthW(hwnd), 4096) + 1)
|
|
self.user32.GetWindowTextW(hwnd, title, len(title))
|
|
return {
|
|
"hwnd": hwnd,
|
|
"pid": pid.value,
|
|
"exe": exe,
|
|
"path": path,
|
|
"title": title.value,
|
|
"visible": bool(self.user32.IsWindowVisible(hwnd)),
|
|
"minimized": bool(self.user32.IsIconic(hwnd)),
|
|
"modal": bool(owner_hwnd and not self.user32.IsWindowEnabled(owner_hwnd)),
|
|
"fingerprint": {
|
|
"class": class_name.value,
|
|
"style": self._popup_get_style(hwnd, -16) & 0x80CF0000,
|
|
"exstyle": self._popup_get_style(hwnd, -20) & 0x08040088,
|
|
"width": round((rect.right - rect.left) * 96 / dpi),
|
|
"height": round((rect.bottom - rect.top) * 96 / dpi),
|
|
"children": children,
|
|
"owner": owner,
|
|
},
|
|
}
|
|
|
|
def _popup_windows(self, paths=None):
|
|
self._init_popup_api()
|
|
windows = []
|
|
process_info = {}
|
|
|
|
def enum_window(hwnd, parameter):
|
|
try:
|
|
if paths is not None:
|
|
pid = ctypes.wintypes.DWORD()
|
|
self.user32.GetWindowThreadProcessId(hwnd, ctypes.byref(pid))
|
|
|
|
if pid.value not in process_info:
|
|
process_info[pid.value] = self.get_exe_info(pid.value)
|
|
|
|
if normalized_path(process_info[pid.value][1]) not in paths:
|
|
return True
|
|
|
|
snapshot = self._popup_window_snapshot(hwnd, process_info)
|
|
|
|
if snapshot:
|
|
windows.append(snapshot)
|
|
except Exception:
|
|
log_exception("PYAS_Popup.PopupWindowMixin._popup_windows.enum_window:246")
|
|
return True
|
|
|
|
return True
|
|
|
|
callback = self._popup_enum_type(enum_window)
|
|
self.user32.EnumWindows(callback, 0)
|
|
return windows
|
|
|
|
def add_popup_rule(self, rule):
|
|
if not isinstance(rule, dict) or rule.get("schema_version") != 2 or not rule.get("id"):
|
|
return False
|
|
|
|
with self.lock_config:
|
|
pending = getattr(self, "_captured_popup_rules", {})
|
|
captured = pending.get(rule["id"])
|
|
|
|
if captured != rule:
|
|
return any(item == rule for item in self.pyas_config.get("block_list", []))
|
|
|
|
target_list = self.pyas_config.setdefault("block_list", [])
|
|
duplicate = next(
|
|
(
|
|
item
|
|
for item in target_list
|
|
if isinstance(item, dict)
|
|
and item.get("schema_version") == 2
|
|
and normalized_path(item.get("path")) == normalized_path(rule.get("path"))
|
|
and item.get("fingerprint") == rule.get("fingerprint")
|
|
and item.get("title") == rule.get("title")
|
|
and item.get("match_mode") == rule.get("match_mode")
|
|
),
|
|
None,
|
|
)
|
|
|
|
if duplicate:
|
|
selected = getattr(self, "_popup_selected", {}).pop(rule["id"], None)
|
|
|
|
if selected:
|
|
self._popup_selected[duplicate["id"]] = selected
|
|
else:
|
|
target_list.append(copy.deepcopy(captured))
|
|
self.save_config()
|
|
self.write_log(
|
|
"INFO",
|
|
"Config Update",
|
|
detail=f"Popup rule added: {rule['exe']} [{rule['id']}]",
|
|
)
|
|
|
|
pending.pop(rule["id"], None)
|
|
return True
|
|
|
|
def _popup_matches(self, rule, target, windows):
|
|
with self.lock_config:
|
|
selected = getattr(self, "_popup_selected", {}).get(rule.get("id"))
|
|
|
|
if (
|
|
selected
|
|
and selected["hwnd"] == target.get("hwnd")
|
|
and selected["pid"] == target.get("pid")
|
|
):
|
|
if self.user32.GetPropW(selected["hwnd"], "PYAS_Popup_Selected") == selected["token"]:
|
|
return normalized_path(rule.get("path")) == normalized_path(target.get("path"))
|
|
|
|
return popup_rule_matches(rule, target, windows)
|
|
|
|
def remove_popup_rules(self, identifiers):
|
|
with self.lock_config:
|
|
rules = self.pyas_config.get("block_list", [])
|
|
identifiers = set(identifiers or [])
|
|
remaining = [
|
|
rule
|
|
for index, rule in enumerate(rules)
|
|
if (rule.get("id") or f"legacy:{index}") not in identifiers
|
|
]
|
|
|
|
if len(remaining) == len(rules):
|
|
return False
|
|
|
|
selected = getattr(self, "_popup_selected", {})
|
|
|
|
for identifier in identifiers:
|
|
record = selected.pop(identifier, None)
|
|
|
|
if (
|
|
record
|
|
and self.user32.GetPropW(record["hwnd"], "PYAS_Popup_Selected")
|
|
== record["token"]
|
|
):
|
|
self.user32.RemovePropW(record["hwnd"], "PYAS_Popup_Selected")
|
|
|
|
self.pyas_config["block_list"] = remaining
|
|
self.write_log(
|
|
"INFO",
|
|
"Config Update",
|
|
detail=f"Popup rules removed: {len(rules) - len(remaining)}",
|
|
)
|
|
self.save_config()
|
|
return True
|
|
|
|
def _restore_popup_windows(self, hidden, active_ids):
|
|
for hwnd, record in list(hidden.items()):
|
|
if self.user32.GetPropW(hwnd, "PYAS_Popup_Blocker") != record["token"]:
|
|
hidden.pop(hwnd, None)
|
|
elif record["rule_id"] not in active_ids:
|
|
self.user32.RemovePropW(hwnd, "PYAS_Popup_Blocker")
|
|
self.user32.ShowWindowAsync(hwnd, 4)
|
|
hidden.pop(hwnd, None)
|
|
|
|
def _hide_popup_window(self, rule, target, windows, hidden):
|
|
hwnd = target["hwnd"]
|
|
current = self._popup_window_snapshot(hwnd)
|
|
|
|
if (
|
|
not current
|
|
or current["pid"] != target["pid"]
|
|
or not self._popup_matches(rule, current, windows)
|
|
):
|
|
return False
|
|
|
|
record = hidden.get(hwnd)
|
|
new_record = (
|
|
not record or self.user32.GetPropW(hwnd, "PYAS_Popup_Blocker") != record["token"]
|
|
)
|
|
|
|
if new_record:
|
|
token = (time.monotonic_ns() & 0x7FFFFFFF) or 1
|
|
|
|
if not self.user32.SetPropW(hwnd, "PYAS_Popup_Blocker", token):
|
|
return False
|
|
|
|
record = {"token": token, "rule_id": rule["id"]}
|
|
hidden[hwnd] = record
|
|
|
|
if not self.user32.ShowWindowAsync(hwnd, 0):
|
|
if new_record:
|
|
self.user32.RemovePropW(hwnd, "PYAS_Popup_Blocker")
|
|
hidden.pop(hwnd, None)
|
|
|
|
return False
|
|
|
|
return True
|
|
|
|
def popup_intercept_thread(self):
|
|
hidden = {}
|
|
last_warning = {}
|
|
|
|
while True:
|
|
try:
|
|
time.sleep(0.5)
|
|
self._init_popup_api()
|
|
|
|
with self.lock_config:
|
|
rules = copy.deepcopy(self.pyas_config.get("block_list", []))
|
|
|
|
valid_rules = []
|
|
|
|
for index, rule in enumerate(rules):
|
|
if isinstance(rule, dict) and (
|
|
rule.get("schema_version") == 2
|
|
and rule.get("id")
|
|
or rule.get("exe")
|
|
and rule.get("class")
|
|
and rule.get("title")
|
|
):
|
|
rule.setdefault("id", f"legacy:{index}")
|
|
valid_rules.append(rule)
|
|
|
|
self._restore_popup_windows(hidden, {rule["id"] for rule in valid_rules})
|
|
|
|
if not valid_rules:
|
|
continue
|
|
|
|
paths = {normalized_path(rule.get("path")) for rule in valid_rules}
|
|
windows = self._popup_windows(paths if all(paths) else None)
|
|
|
|
for target in windows:
|
|
if not target.get("visible") or target.get("minimized"):
|
|
continue
|
|
|
|
for rule in valid_rules:
|
|
if self._popup_matches(rule, target, windows):
|
|
first_hide = target["hwnd"] not in hidden
|
|
|
|
if self._hide_popup_window(rule, target, windows, hidden):
|
|
if first_hide:
|
|
self.write_log(
|
|
"INFO",
|
|
"Popup Blocker",
|
|
source=target["path"],
|
|
detail=f"Hidden popup window: {target['title']}",
|
|
pid=target["pid"],
|
|
)
|
|
elif first_hide:
|
|
key = (rule["id"], target["hwnd"], target["pid"])
|
|
now = time.monotonic()
|
|
|
|
if now - last_warning.get(key, -30) >= 30:
|
|
self.write_log(
|
|
"WARN",
|
|
"Popup Blocker",
|
|
source=target["path"],
|
|
detail="Unable to hide selected window",
|
|
pid=target["pid"],
|
|
success=False,
|
|
)
|
|
last_warning[key] = now
|
|
|
|
break
|
|
except Exception as e:
|
|
log_exception("PYAS_Popup.PopupWindowMixin.popup_intercept_thread:369")
|
|
now = time.monotonic()
|
|
|
|
if now - last_warning.get("error", -30) >= 30:
|
|
self.write_log("WARN", "Popup Blocker", detail=str(e), success=False)
|
|
last_warning["error"] = now
|