ghostwire/whatsapp_suite/desktop.py

"""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