130 lines
6.2 KiB
Python
130 lines
6.2 KiB
Python
"""Shared verified Desktop process and window discovery."""
|
|
import ctypes
|
|
import hashlib
|
|
import re
|
|
import warnings
|
|
from pathlib import Path
|
|
TESTED_BUILD = "2.2639.100.0"
|
|
SUPPORTED_BUILD = TESTED_BUILD # Backward-compatible constant; not an allowlist.
|
|
CORE_SHA256 = "c6da3c67eda66a8d48ff2ae11f68085f1f8250111138a6ae384d38eefb4da536"
|
|
|
|
def check_core_build(path):
|
|
"""Describe tested-build compatibility without rejecting other versions.
|
|
|
|
Native object/method checks still determine whether the bridge can operate.
|
|
File access failures remain errors; a hash mismatch alone is advisory.
|
|
"""
|
|
path = Path(path)
|
|
match = re.search(r'WhatsAppDesktop_([0-9.]+)_', str(path), re.IGNORECASE)
|
|
actual_build = match.group(1) if match else None
|
|
actual_hash = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
matched = actual_build == TESTED_BUILD and actual_hash == CORE_SHA256
|
|
return {'tested_build': TESTED_BUILD, 'detected_build': actual_build,
|
|
'tested_core_sha256': CORE_SHA256, 'detected_core_sha256': actual_hash,
|
|
'matches_tested_configuration': matched,
|
|
'warning': None if matched else
|
|
'Untested WhatsApp/WebView configuration; attempting the bridge. '
|
|
'Native object checks remain enabled. Compatibility is not verified; '
|
|
'changed native layouts may cause failure or process instability.'}
|
|
|
|
def visible_ui_threads(pid):
|
|
"""Inspect window ownership; no clicks, navigation or window messages."""
|
|
import ctypes
|
|
from ctypes import wintypes
|
|
user32 = ctypes.WinDLL('user32', use_last_error=True)
|
|
user32.GetWindowThreadProcessId.argtypes = [wintypes.HWND, ctypes.POINTER(wintypes.DWORD)]
|
|
user32.GetWindowThreadProcessId.restype = wintypes.DWORD
|
|
user32.IsWindowVisible.argtypes = [wintypes.HWND]
|
|
user32.IsWindowVisible.restype = wintypes.BOOL
|
|
callback_type = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.HWND, wintypes.LPARAM)
|
|
threads = set()
|
|
def visit(hwnd, parameter):
|
|
process = wintypes.DWORD()
|
|
thread = user32.GetWindowThreadProcessId(hwnd, ctypes.byref(process))
|
|
if process.value == pid and user32.IsWindowVisible(hwnd):
|
|
threads.add(thread)
|
|
return True
|
|
callback = callback_type(visit)
|
|
user32.EnumWindows.argtypes = [callback_type, wintypes.LPARAM]
|
|
user32.EnumWindows.restype = wintypes.BOOL
|
|
if not user32.EnumWindows(callback, 0):
|
|
raise ctypes.WinError(ctypes.get_last_error())
|
|
return threads
|
|
|
|
def detect_desktop_pid(requested_pid=None, compatibility=None):
|
|
"""Find the running Desktop host using Python and Windows APIs."""
|
|
import ctypes
|
|
from ctypes import wintypes as w
|
|
import hashlib
|
|
class ProcessEntry(ctypes.Structure):
|
|
_fields_ = [('dwSize',w.DWORD),('cntUsage',w.DWORD),('th32ProcessID',w.DWORD),
|
|
('th32DefaultHeapID',ctypes.c_size_t),('th32ModuleID',w.DWORD),
|
|
('cntThreads',w.DWORD),('th32ParentProcessID',w.DWORD),
|
|
('pcPriClassBase',w.LONG),('dwFlags',w.DWORD),('szExeFile',w.WCHAR*260)]
|
|
class ModuleEntry(ctypes.Structure):
|
|
_fields_ = [('dwSize',w.DWORD),('th32ModuleID',w.DWORD),('th32ProcessID',w.DWORD),
|
|
('GlblcntUsage',w.DWORD),('ProccntUsage',w.DWORD),
|
|
('modBaseAddr',ctypes.POINTER(w.BYTE)),('modBaseSize',w.DWORD),
|
|
('hModule',w.HMODULE),('szModule',w.WCHAR*256),('szExePath',w.WCHAR*260)]
|
|
kernel = ctypes.WinDLL('kernel32',use_last_error=True)
|
|
kernel.CreateToolhelp32Snapshot.argtypes = [w.DWORD,w.DWORD]
|
|
kernel.CreateToolhelp32Snapshot.restype = w.HANDLE
|
|
kernel.CloseHandle.argtypes = [w.HANDLE]
|
|
kernel.CloseHandle.restype = w.BOOL
|
|
for name, entry in [('Process32FirstW',ProcessEntry),('Process32NextW',ProcessEntry),
|
|
('Module32FirstW',ModuleEntry),('Module32NextW',ModuleEntry)]:
|
|
fn = getattr(kernel,name)
|
|
fn.argtypes = [w.HANDLE,ctypes.POINTER(entry)]
|
|
fn.restype = w.BOOL
|
|
invalid = ctypes.c_void_p(-1).value
|
|
snapshot = kernel.CreateToolhelp32Snapshot(2,0)
|
|
if snapshot == invalid:
|
|
raise ctypes.WinError(ctypes.get_last_error())
|
|
roots = []
|
|
try:
|
|
entry = ProcessEntry(dwSize=ctypes.sizeof(ProcessEntry))
|
|
exists = kernel.Process32FirstW(snapshot,ctypes.byref(entry))
|
|
while exists:
|
|
if entry.szExeFile.casefold() == 'whatsapp.root.exe':
|
|
roots.append(entry.th32ProcessID)
|
|
exists = kernel.Process32NextW(snapshot,ctypes.byref(entry))
|
|
finally:
|
|
kernel.CloseHandle(snapshot)
|
|
candidates = []
|
|
rejected = []
|
|
for pid in roots:
|
|
if requested_pid is not None and pid != requested_pid:
|
|
continue
|
|
snapshot = kernel.CreateToolhelp32Snapshot(8|16,pid)
|
|
if snapshot == invalid:
|
|
continue
|
|
try:
|
|
entry = ModuleEntry(dwSize=ctypes.sizeof(ModuleEntry))
|
|
exists = kernel.Module32FirstW(snapshot,ctypes.byref(entry))
|
|
while exists:
|
|
if entry.szModule.casefold() == 'microsoft.web.webview2.core.dll':
|
|
path = Path(entry.szExePath)
|
|
if len(visible_ui_threads(pid)) == 1:
|
|
try:
|
|
details = check_core_build(path)
|
|
candidates.append((pid, details))
|
|
except (ValueError, OSError) as error:
|
|
rejected.append(str(error))
|
|
break
|
|
exists = kernel.Module32NextW(snapshot,ctypes.byref(entry))
|
|
except OSError:
|
|
continue
|
|
finally:
|
|
kernel.CloseHandle(snapshot)
|
|
if not candidates:
|
|
raise ValueError('; '.join(rejected) if rejected else 'No running WhatsApp Desktop host with one identifiable UI thread found; start WhatsApp normally')
|
|
if len(candidates) != 1:
|
|
raise ValueError('Multiple active Desktop hosts found; specify --pid: '+', '.join(str(item[0]) for item in candidates))
|
|
selected, details = candidates[0]
|
|
if compatibility is not None:
|
|
compatibility.update(details)
|
|
if details['warning']:
|
|
warnings.warn(details['warning'], RuntimeWarning, stacklevel=2)
|
|
return selected
|
|
|