"""Standalone fixed arithmetic and capability tests in the WhatsApp WebView. Windows x64 and 64-bit Python; standard library only. The bundled native DLL is hash-verified and cached in the workspace. Capability tests use fixed queries, empty-input reads and simple getters. Explicit message-json and contacts suites serialize the fixed search results or a five-contact sample. There is no arbitrary JavaScript CLI and settings setters are never invoked. """ from __future__ import annotations import argparse import ctypes from ctypes import wintypes as w import hashlib import json import math import os from pathlib import Path import tempfile import time import uuid import sys from .common import BASE, verified_helper from .desktop import visible_ui_threads, detect_desktop_pid, TESTED_BUILD WAKE = 0x8000 + 0x498 PROTOCOL = 0x4d415432 # SHA-256 of assets/math_probe.dll. EMBEDDED_SHA256 = 'bdcf0449d35102963c35f0f8ba985ffe3e5f83c083da5587b4978c82911c555f' CASES = ( ('(17 * 23) - 19', 372), ('2 ** 10', 1024), ('Math.sqrt(144) + Math.cbrt(27)', 15), ('Math.hypot(3, 4)', 5), ('Math.sin(Math.PI / 2)', 1), ('[1,2,3,4,5].reduce((sum,n) => sum + n*n, 0)', 55), ('0.1 + 0.2', 0.3), ('(12345678901234567890n * 9n).toString()', '111111110111111111010'), ) # Fixed probes only. Availability checks never invoke the selected member. CAPABILITIES = ( ('search.entry', 'search', 'WAWebMsgCollection', 'MsgCollection.search', 'search'), ('search.local', 'search', 'WAWebDBMessageFindLocal', 'msgFindSearch', 'presence'), ('search.fts', 'search', 'WAWebFtsClient', 'ftsClient.search', 'presence'), ('messages.by_id', 'messages', 'WAWebMsgCollection', 'MsgCollection.getMessagesById', 'empty_messages'), ('messages.hydrate', 'messages', 'WAWebMsgCollection', 'MsgCollection.hydrateOrGetMessages', 'empty_array'), ('messages.before', 'messages', 'WAWebDBMessageFindLocal', 'msgFindBefore', 'presence'), ('messages.after', 'messages', 'WAWebDBMessageFindLocal', 'msgFindAfter', 'presence'), ('messages.media', 'messages', 'WAWebDBMessageFindLocal', 'msgFindMedia', 'presence'), ('messages.serialization', 'messages', 'WAWebDBMessageSerialization', 'messageFromDbRow', 'presence'), ('storage.message_table', 'storage', 'WAWebSchemaMessage', 'getMessageTable', 'table_empty'), ('storage.indexeddb', 'storage', 'WAWebIndexedDB', 'default.open', 'presence'), ('storage.class', 'storage', 'WAWebStorage', 'Storage', 'presence'), ('storage.permanent', 'storage', 'WAWebPermanentStorage', 'default.getItem', 'presence'), ('settings.theme', 'settings', 'WAWebUserPrefsGeneral', 'getTheme', 'theme'), ('settings.theme_setter', 'settings', 'WAWebUserPrefsGeneral', 'setTheme', 'presence'), ('settings.dark_theme', 'settings', 'WAWebThemeSession', 'getIsDarkTheme', 'boolean'), ('settings.notifications', 'settings', 'WAWebUserPrefsNotifications', 'getGlobalNotifications', 'flag'), ('settings.notifications_setter', 'settings', 'WAWebUserPrefsNotifications', 'setGlobalNotifications', 'presence'), ('settings.privacy', 'settings', 'WAWebUserPrefsGeneral', 'getUserPrivacySettings', 'privacy_shape'), ('settings.auto_download', 'settings', 'WAWebUserPrefsGeneral', 'getAutoDownloadPhotos', 'boolean'), ('settings.archive', 'settings', 'WAWebSettingsGetters', 'getArchive', 'presence'), ('settings.apply', 'settings', 'WAWebSettingsBridgeApi', 'SettingsBridgeApi.applyAppSetting', 'presence'), ('settings.sync', 'settings', 'WAWebSettingsSyncBridge', 'sendSettingChange', 'presence'), ('chats.collection', 'chats', 'WAWebChatCollection', 'ChatCollection.get', 'presence'), ('contacts.collection', 'contacts', 'WAWebContactCollection', 'ContactCollection.get', 'presence'), ('bridge.ready', 'bridge', 'WAWebWindowsHybridBridgeFactory', 'getIsBridgeReady', 'boolean'), ('bridge.get', 'bridge', 'WAWebWindowsHybridBridgeFactory', 'getWindowsBridge', 'presence'), ('bridge.frontend', 'bridge', 'WAWebBackendApi', 'frontendSendAndReceive', 'presence'), ('bridge.worker', 'bridge', 'WAWebWorkerSafeBackendApi', 'workerSafeSendAndReceive', 'presence'), ) def capability_expressions(probe_id): key = '__whatsapp_fixed_capability_' + probe_id specs = json.dumps(CAPABILITIES, separators=(',', ':')) start = r'''(() => { const key=__KEY__,id=__ID__,specs=__SPECS__; const state={probe_id:id,state:'pending',checks:[]}; window[key]=state; if(typeof require!=='function'){ state.state='error';state.error='WhatsApp module loader unavailable';return {probe_id:id,state:'started'}; } async function bounded(fn){ let timer; try{return await Promise.race([Promise.resolve().then(fn),new Promise((_,reject)=>{ timer=setTimeout(()=>reject(Error('Fixed getter/query timeout')),2500); })]);}finally{clearTimeout(timer);} } (async()=>{ const loaded=new Map(); for(const [test,area,module,path,action] of specs){ const row={id:test,area,module,member:path,probe:action,invoked:false,passed:false}; try{ let exports=loaded.get(module); if(!exports){exports=require(module);loaded.set(module,exports);} const parts=path.split('.');let owner=exports; // Meta's global require may unwrap a default-only module. if(parts[0]==='default'&&parts.length>1&&exports?.default===undefined&& typeof exports?.[parts[parts.length-1]]==='function'){ parts.shift();row.import_shape='default-export-unwrapped'; } for(const part of parts.slice(0,-1))owner=owner?.[part]; const value=owner?.[parts[parts.length-1]]; row.available=typeof value==='function'; if(!row.available){ row.status='unavailable'; row.summary={module_type:typeof exports,default_type:typeof exports?.default, direct_member_type:typeof exports?.[parts[parts.length-1]], export_names:exports&&typeof exports==='object'?Object.keys(exports).slice(0,12):[]}; state.checks.push(row);continue; } if(action==='presence'){row.passed=true;row.status='available';} else{ row.invoked=true; if(action==='search'){ const result=await bounded(()=>value.call(owner,'meow',1,3)); row.summary={query:'meow',matches:Array.isArray(result?.messages)?result.messages.length:null, canceled:result?.canceled===true,eof:result?.eof===true}; row.passed=Array.isArray(result?.messages)&&result?.canceled!==true; }else if(action==='empty_messages'||action==='empty_array'){ const result=await bounded(()=>value.call(owner,[])); const list=action==='empty_array'?result:result?.messages; row.summary={input_count:0,returned_count:Array.isArray(list)?list.length:null}; row.passed=Array.isArray(list)&&list.length===0; }else if(action==='table_empty'){ const table=value.call(owner); row.summary={methods:['get','bulkGet','all','count'].filter(name=>typeof table?.[name]==='function')}; if(typeof table?.bulkGet!=='function')throw Error('Message table bulkGet unavailable in this context'); const result=await bounded(()=>table.bulkGet([])); row.summary.input_count=0;row.summary.returned_count=Array.isArray(result)?result.length:null; row.passed=Array.isArray(result)&&result.length===0; }else{ const result=await bounded(()=>value.call(owner)); if(action==='privacy_shape'){ row.summary={type:typeof result,fields:result&&typeof result==='object'?Object.keys(result).sort():[]}; row.passed=result!==null&&typeof result==='object'&&!Array.isArray(result); }else if(action==='theme'){ row.summary={type:typeof result}; if(['light','dark','system','auto'].includes(result))row.summary.value=result; row.passed=typeof result==='string'; }else{ row.summary={type:typeof result}; row.passed=action==='boolean'?typeof result==='boolean': result==null||typeof result==='boolean'||typeof result==='number'; } } row.status=row.passed?'passed':'unexpected-result'; } }catch(error){row.status='error';row.error=String(error).slice(0,400);} state.checks.push(row); } state.state='done'; })().catch(error=>{state.state='error';state.error=String(error).slice(0,400);}); return {probe_id:id,state:'started'}; })()'''.replace('__KEY__', json.dumps(key)).replace('__ID__', json.dumps(probe_id)).replace('__SPECS__', specs) poll = '''(() => {const key=__KEY__,state=window[key]; if(!state)return {probe_id:__ID__,state:'error',error:'Probe state missing'}; if(state.state==='done'||state.state==='error')delete window[key]; return state;})()'''.replace('__KEY__', json.dumps(key)).replace('__ID__', json.dumps(probe_id)) remove = 'delete window[' + json.dumps(key) + ']' if len(start.encode('utf-16-le')) // 2 >= 8192: raise ValueError('Fixed capability probe exceeds native input buffer') return start, poll, remove def validate_capabilities(parsed, probe_id): if parsed.get('probe_id') != probe_id or parsed.get('state') != 'done': raise ValueError('Capability probe incomplete: ' + str(parsed.get('error', parsed.get('state')))) checks = parsed.get('checks', []) if [row.get('id') for row in checks] != [spec[0] for spec in CAPABILITIES]: raise ValueError('Capability response does not match the fixed test list') return checks def data_expressions(probe_id, kind): """Fixed meow search or loaded-contact sample; no arbitrary query/code input.""" key = '__whatsapp_fixed_data_' + probe_id start = r'''(() => { const key=__KEY__,id=__ID__,kind=__KIND__; const state={probe_id:id,state:'pending'};window[key]=state; function snapshot(model){ if(typeof model?.serialize!=='function')throw Error('App model serializer unavailable'); return model.serialize(); } Promise.resolve().then(async()=>{ if(typeof require!=='function')throw Error('WhatsApp module loader unavailable'); let payload; if(kind==='message-json'){ const collection=require('WAWebMsgCollection').MsgCollection; const result=await collection.search('meow',1,3); if(!Array.isArray(result?.messages)||result?.canceled)throw Error('Search did not complete'); payload={probe_id:id,kind,query:'meow',matched_count:result.messages.length, eof:result.eof===true,serializer:'BaseModel.serialize',messages:result.messages.map(snapshot)}; }else if(kind==='history-read'){ payload={probe_id:id,kind,...await __HISTORY_PROBE__}; }else{ const collection=require('WAWebContactCollection').ContactCollection; if(typeof collection?.getModelsArray!=='function')throw Error('Contact getter unavailable'); const contacts=collection.getModelsArray(); if(!Array.isArray(contacts))throw Error('Contact getter returned an unexpected type'); payload={probe_id:id,kind,loaded_count:contacts.length,sample_limit:5, serializer:'BaseModel.serialize',contacts:contacts.slice(0,5).map(snapshot)}; } const seen=new WeakMap();let reference=0; state.json=JSON.stringify(payload,function(name,value){ if(value===undefined)return {$type:'undefined'}; if(typeof value==='bigint')return {$type:'BigInt',decimal:value.toString()}; if(typeof value==='function')return {$type:'function',name:value.name}; if(value&&typeof value==='object'){ if(seen.has(value))return {$ref:seen.get(value)}; seen.set(value,++reference); if(value instanceof ArrayBuffer)return {$type:'ArrayBuffer',bytes:Array.from(new Uint8Array(value))}; if(ArrayBuffer.isView(value))return {$type:value.constructor.name,bytes:Array.from(new Uint8Array(value.buffer,value.byteOffset,value.byteLength))}; if(value instanceof Map)return {$type:'Map',entries:Array.from(value.entries())}; if(value instanceof Set)return {$type:'Set',values:Array.from(value.values())}; } return value; }); state.state='ready'; }).catch(error=>{state.state='error';state.error=String(error).slice(0,400);}); return {probe_id:id,state:'started'}; })()'''.replace('__KEY__', json.dumps(key)).replace('__ID__', json.dumps(probe_id)).replace('__KIND__', json.dumps(kind)) from .history_read import PROBE as history_probe start = start.replace('__HISTORY_PROBE__', history_probe) poll = '''(() => {const value=window[__KEY__];if(!value)return {probe_id:__ID__,state:'error',error:'Data probe state missing'}; return {probe_id:value.probe_id,state:value.state,error:value.error,length:value.json?.length};})()'''.replace('__KEY__', json.dumps(key)).replace('__ID__', json.dumps(probe_id)) return start, poll, key, 'delete window[' + json.dumps(key) + ']' def data_chunk_expression(key, offset): return '''(() => {const value=window[__KEY__];if(value?.state!=='ready')throw Error('Data not ready'); const start=__OFFSET__;let end=Math.min(start+4096,value.json.length); if(end=0xd800&&value.json.charCodeAt(end-1)<=0xdbff)end++; return {probe_id:value.probe_id,offset:start,next:end,chunk:value.json.slice(start,end)};})()'''.replace('__KEY__', json.dumps(key)).replace('__OFFSET__', str(offset)) def extract_helper(): return verified_helper('math_probe.dll', EMBEDDED_SHA256, '.whatsapp-math-native', 'math_probe.dll') def fixed_expression(probe_id): values = ','.join('{expression:' + json.dumps(expression) + ',value:(' + expression + ')}' for expression, _ in CASES) return '({probe_id:' + json.dumps(probe_id) + ',cases:[' + values + ']})' def validate_result(raw, probe_id): parsed = json.loads(raw) if parsed.get('probe_id') != probe_id or len(parsed.get('cases', [])) != len(CASES): raise ValueError('Arithmetic response does not match this run') results = [] for (expression, expected), observed in zip(CASES, parsed['cases']): value = observed.get('value') passed = observed.get('expression') == expression if isinstance(expected, float): passed = passed and type(value) in (int, float) and math.isclose( value, expected, rel_tol=1e-12, abs_tol=1e-12) else: passed = passed and type(value) is type(expected) and value == expected results.append(dict(expression=expression, expected=expected, actual=value, passed=passed)) return results class Packet(ctypes.Structure): _fields_ = [('state', w.LONG), ('hr', w.LONG), ('thread', w.DWORD), ('protocol', w.DWORD), ('script', w.WCHAR * 8192), ('result', w.WCHAR * 32768)] def windows_api(): kernel = ctypes.WinDLL('kernel32', use_last_error=True) user = ctypes.WinDLL('user32', use_last_error=True) declarations = { 'CreateFileMappingW': ([w.HANDLE, ctypes.c_void_p, w.DWORD, w.DWORD, w.DWORD, w.LPCWSTR], w.HANDLE), 'MapViewOfFile': ([w.HANDLE, w.DWORD, w.DWORD, w.DWORD, ctypes.c_size_t], ctypes.c_void_p), 'UnmapViewOfFile': ([ctypes.c_void_p], w.BOOL), 'CloseHandle': ([w.HANDLE], w.BOOL), 'LoadLibraryW': ([w.LPCWSTR], w.HMODULE), 'FreeLibrary': ([w.HMODULE], w.BOOL), 'GetProcAddress': ([w.HMODULE, ctypes.c_char_p], ctypes.c_void_p), 'OpenProcess': ([w.DWORD, w.BOOL, w.DWORD], w.HANDLE), 'WaitForSingleObject': ([w.HANDLE, w.DWORD], w.DWORD), } for name, (arguments, returns) in declarations.items(): function = getattr(kernel, name) function.argtypes = arguments function.restype = returns user.SetWindowsHookExW.argtypes = [ctypes.c_int, ctypes.c_void_p, w.HMODULE, w.DWORD] user.SetWindowsHookExW.restype = w.HANDLE user.UnhookWindowsHookEx.argtypes = [w.HANDLE] user.UnhookWindowsHookEx.restype = w.BOOL user.PostThreadMessageW.argtypes = [w.DWORD, w.UINT, w.WPARAM, w.LPARAM] user.PostThreadMessageW.restype = w.BOOL return kernel, user def wait_for_state(packet, desired, seconds, wake, alive, phase): deadline = time.monotonic() + seconds while time.monotonic() < deadline: if not alive(): raise RuntimeError('WhatsApp exited during ' + phase) wake() time.sleep(.025) if packet.state == desired: return if packet.state == -1: raise RuntimeError(phase + ': native error 0x' + format(packet.hr & 0xffffffff, '08x')) if phase == 'startup': raise TimeoutError('Helper startup not acknowledged within ' + str(seconds) + 's; no JavaScript requested. Check DLL path access in packaged WhatsApp.') raise TimeoutError(phase + ' timed out (native state ' + str(packet.state) + ')') def helper_loaded(pid, path): """Observe Windows' asynchronous hook-DLL unloading using module metadata.""" class ModuleEntry(ctypes.Structure): _fields_ = [('dwSize', w.DWORD), ('th32ModuleID', w.DWORD), ('th32ProcessID', w.DWORD), ('GlblcntUsage', w.DWORD), ('ProccntUsage', w.DWORD), ('modBaseAddr', ctypes.c_void_p), ('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 in ('Module32FirstW', 'Module32NextW'): function = getattr(kernel, name) function.argtypes = [w.HANDLE, ctypes.POINTER(ModuleEntry)] function.restype = w.BOOL snapshot = kernel.CreateToolhelp32Snapshot(8 | 16, pid) if snapshot == ctypes.c_void_p(-1).value: raise ctypes.WinError(ctypes.get_last_error()) try: entry = ModuleEntry(dwSize=ctypes.sizeof(ModuleEntry)) present = kernel.Module32FirstW(snapshot, ctypes.byref(entry)) while present: if os.path.normcase(entry.szExePath) == os.path.normcase(str(path)): return True present = kernel.Module32NextW(snapshot, ctypes.byref(entry)) error = ctypes.get_last_error() if error != 18: # ERROR_NO_MORE_FILES raise ctypes.WinError(error) return False finally: kernel.CloseHandle(snapshot) def run_math(pid=None, suite='math', _operation=None): scopes = {'math': 'fixed-arithmetic-only', 'capabilities': 'fixed-capability-probes', 'all': 'fixed-arithmetic-and-capability-probes', 'message-json': 'fixed-meow-message-models', 'contacts': 'loaded-contact-sample', 'history-read': 'oldest-local-messages-through-javascript-no-cdp', 'history-export': 'all-local-message-rows-through-javascript-no-cdp', 'send-self': 'one-text-message-to-own-account-through-javascript', 'contacts-export': 'all-local-contact-records-through-javascript', 'webhook': 'explicit-javascript-message-listener-management'} report = {'passed': False, 'scope': scopes.get(suite, 'invalid-suite'), 'suite': suite, 'frida_used': False, 'startup_acknowledged': False, 'javascript_executed': False, 'cleanup': {'acknowledged': False, 'hook_removed': False}, 'tested_build': TESTED_BUILD} kernel = user = None module = hook = mapping = view = process = None packet = None remove_expression = None execute = None try: if suite not in scopes: raise ValueError('Unknown suite; choose math, capabilities, all, message-json or contacts') if os.name != 'nt' or ctypes.sizeof(ctypes.c_void_p) != 8: raise ValueError('Windows and 64-bit Python are required') compatibility = {} selected = detect_desktop_pid(pid, compatibility=compatibility) report['compatibility'] = compatibility thread = next(iter(visible_ui_threads(selected))) report.update(pid=selected, ui_thread_id=thread, pid_auto_detected=pid is None) kernel, user = windows_api() process = kernel.OpenProcess(0x100000, False, selected) # SYNCHRONIZE only if not process: raise ctypes.WinError(ctypes.get_last_error()) dll, cache = extract_helper() report.update(native_helper_path=str(dll), cache=cache) ctypes.set_last_error(0) mapping = kernel.CreateFileMappingW(w.HANDLE(-1), None, 4, 0, ctypes.sizeof(Packet), 'Local\\WhatsAppMathProbe-v2-' + str(selected)) mapping_error = ctypes.get_last_error() if not mapping: raise ctypes.WinError(mapping_error) if mapping_error == 183: raise ValueError('Another math probe owns this WhatsApp instance; no hook installed') view = kernel.MapViewOfFile(mapping, 0xf001f, 0, 0, ctypes.sizeof(Packet)) if not view: raise ctypes.WinError(ctypes.get_last_error()) packet = Packet.from_address(view) packet.thread = thread packet.protocol = PROTOCOL packet.state = 3 # Startup only: native helper must acknowledge before any JS. module = kernel.LoadLibraryW(str(dll)) if not module: raise ctypes.WinError(ctypes.get_last_error()) callback = kernel.GetProcAddress(module, b'ProbeHook') if not callback: raise ctypes.WinError(ctypes.get_last_error()) hook = user.SetWindowsHookExW(3, callback, module, thread) if not hook: raise ctypes.WinError(ctypes.get_last_error()) def wake(): if not user.PostThreadMessageW(thread, WAKE, 0, 0): raise ctypes.WinError(ctypes.get_last_error()) def alive(): status = kernel.WaitForSingleObject(process, 0) if status == 0xffffffff: raise ctypes.WinError(ctypes.get_last_error()) return status == 258 started = time.monotonic() wait_for_state(packet, 4, 5, wake, alive, 'startup') report.update(startup_acknowledged=True, startup_seconds=round(time.monotonic() - started, 3)) probe_id = uuid.uuid4().hex report['probe_id'] = probe_id def execute(expression, phase): if len(expression.encode('utf-16-le')) // 2 >= 8192: raise ValueError('Fixed probe exceeds native input buffer') packet.script = expression packet.result = '' packet.hr = 0 packet.state = 0 wait_for_state(packet, 2, 20, wake, alive, phase) report['javascript_executed'] = True return packet.result if suite in ('history-export', 'send-self', 'contacts-export', 'webhook'): if _operation is None: raise ValueError('Use the export-zip command for history export') _operation(execute, probe_id, report) report['operation_passed'] = True if suite in ('math', 'all'): report['tests'] = validate_result(execute(fixed_expression(probe_id), 'arithmetic'), probe_id) report['math_passed'] = all(result['passed'] for result in report['tests']) if suite in ('capabilities', 'all'): start, poll, remove_expression = capability_expressions(probe_id) initial = json.loads(execute(start, 'capability startup')) if initial.get('probe_id') != probe_id or initial.get('state') != 'started': raise ValueError('Capability startup response does not match this run') deadline = time.monotonic() + 80 while time.monotonic() < deadline: parsed = json.loads(execute(poll, 'capability poll')) if parsed.get('state') != 'pending': checks = validate_capabilities(parsed, probe_id) report['capabilities'] = checks report['capability_summary'] = { 'total': len(checks), 'passed': sum(row['passed'] for row in checks), 'invoked': sum(row['invoked'] for row in checks), 'unavailable': sum(row.get('status') == 'unavailable' for row in checks), 'errors': sum(row.get('status') == 'error' for row in checks), } report['capabilities_passed'] = all(row['passed'] for row in checks) remove_expression = None # Poll deletes completed state in the page. report['page_probe_state_removed'] = True break time.sleep(.1) else: raise TimeoutError('Fixed capability suite did not finish within 80 seconds') if suite in ('message-json', 'contacts', 'history-read'): start, poll, key, remove_expression = data_expressions(probe_id, suite) initial = json.loads(execute(start, 'data startup')) if initial.get('probe_id') != probe_id or initial.get('state') != 'started': raise ValueError('Data probe startup does not match this run') deadline = time.monotonic() + 30 while time.monotonic() < deadline: metadata = json.loads(execute(poll, 'data poll')) if metadata.get('probe_id') != probe_id: raise ValueError('Data probe session mismatch') if metadata.get('state') == 'error': raise ValueError(str(metadata.get('error'))) if metadata.get('state') == 'ready': break time.sleep(.1) else: raise TimeoutError('Fixed data probe did not become ready') length = metadata.get('length') if type(length) is not int or not 0 < length <= 16 * 1024 * 1024: raise ValueError('Data probe exceeds the bounded JSON transport size') pieces = [] offset = 0 while offset < length: part = json.loads(execute(data_chunk_expression(key, offset), 'data chunk')) following = part.get('next') if (part.get('probe_id') != probe_id or part.get('offset') != offset or type(following) is not int or not offset < following <= length or not isinstance(part.get('chunk'), str)): raise ValueError('Invalid JSON transport chunk') if len(part['chunk'].encode('utf-16-le', errors='surrogatepass')) // 2 != following - offset: raise ValueError('JSON transport chunk length mismatch') pieces.append(part['chunk']) offset = following data = json.loads(''.join(pieces)) if data.get('probe_id') != probe_id or data.get('kind') != suite: raise ValueError('JSON payload does not match this data probe') report['data'] = data report['data_passed'] = isinstance(data.get('contacts' if suite == 'contacts' else 'messages'), list) if suite == 'history-read': report['data_passed'] = report['data_passed'] and bool(data['messages']) and all( isinstance(row.get('timestamp'), (int, float)) and row['timestamp'] == row.get('selected_timestamp') for row in data['messages']) report['serialization_notes'] = ( 'Declared model properties and nested collections through app serialize(); ' 'undefined/BigInt/binary values are tagged. Repeated object references use traversal-order $ref IDs. ' 'This is a model snapshot, not prototype methods or every private heap field.' ) execute(remove_expression, 'data state cleanup') remove_expression = None report['page_probe_state_removed'] = True except Exception as error: report['error'] = str(error) finally: cleanup = report['cleanup'] if remove_expression and execute and packet is not None and packet.state in (2, 4, -1): try: execute(remove_expression, 'capability state cleanup') report['page_probe_state_removed'] = True except Exception as error: cleanup['page_state_error'] = str(error) if hook and packet is not None: try: packet.state = 6 wait_for_state(packet, 7, 5, wake, alive, 'cleanup') cleanup['acknowledged'] = True cleanup['async_hresult'] = '0x' + format(packet.hr & 0xffffffff, '08x') if packet.hr < 0: cleanup['error'] = 'Native cancel/close failed: ' + cleanup['async_hresult'] except Exception as error: cleanup['error'] = str(error) cleanup['hook_removed'] = bool(user.UnhookWindowsHookEx(hook)) if not cleanup['hook_removed']: cleanup['unhook_error'] = str(ctypes.WinError(ctypes.get_last_error())) elif not hook: cleanup['not_started'] = True for name, handle, function in [('module', module, 'FreeLibrary'), ('mapping_view', view, 'UnmapViewOfFile'), ('mapping_handle', mapping, 'CloseHandle'), ('process_handle', process, 'CloseHandle')]: if handle: ok = bool(getattr(kernel, function)(handle)) cleanup[name + '_released'] = ok if not ok: cleanup[name + '_error'] = str(ctypes.WinError(ctypes.get_last_error())) if hook and cleanup['hook_removed']: try: deadline = time.monotonic() + 3 while helper_loaded(selected, dll) and time.monotonic() < deadline: time.sleep(.05) cleanup['native_helper_unloaded'] = not helper_loaded(selected, dll) if not cleanup['native_helper_unloaded']: cleanup['unload_error'] = 'Windows has not unloaded the hook DLL yet' except Exception as error: cleanup['unload_error'] = str(error) suites_ok = ((suite not in ('math', 'all') or report.get('math_passed', False)) and (suite not in ('capabilities', 'all') or report.get('capabilities_passed', False)) and (suite not in ('message-json', 'contacts', 'history-read') or report.get('data_passed', False)) and (suite not in ('history-export', 'send-self', 'contacts-export') or report.get('operation_passed', False))) report['passed'] = (suites_ok and 'error' not in report and cleanup['acknowledged'] and cleanup['hook_removed'] and not any( key.endswith('error') for key in cleanup)) return report def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--suite', choices=('math', 'capabilities', 'all', 'message-json', 'contacts', 'history-read'), default='math', help='Fixed tests (default: math); message-json prints meow matches, contacts prints a five-contact sample') parser.add_argument('--pid', type=int, help='Optional Desktop host PID') parser.add_argument('--output', type=Path) args = parser.parse_args(argv) if args.output is None: args.output = BASE / ('whatsapp-' + args.suite + '-report.json') report = run_math(args.pid, args.suite) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, indent=2) + '\n', encoding='utf-8') print(json.dumps(report, indent=2)) return 0 if report['passed'] else 1 if __name__ == '__main__': raise SystemExit(main())