593 lines
32 KiB
Python
593 lines
32 KiB
Python
"""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<value.json.length&&value.json.charCodeAt(end-1)>=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())
|