ghostwire/whatsapp_suite/native_search.py

"""Bounded live native WebView search backend."""
import os
import json
from pathlib import Path
from .common import BASE, verified_helper
from .desktop import detect_desktop_pid, visible_ui_threads


EMBEDDED_PROBE_SHA256 = '911beb9f6bb52390cde8a75562ca1d6c46f4629f641426a5250286903de60895'


def build_native_probe():
    """Extract the bundled Windows x64 helper; no compiler or downloads."""
    return verified_helper('search_probe.dll', EMBEDDED_PROBE_SHA256, '.whatsapp-native', 'probe.dll')[0]


def native_webview_test(pid, output, config, search_query=None, limit=50):
    """Python/Windows backend; no Frida import, server, agent or runtime."""
    import ctypes
    from ctypes import wintypes as w
    import time
    import hashlib
    automatic_pid = pid is None
    if automatic_pid:
        pid = detect_desktop_pid()
    threads = visible_ui_threads(pid)
    if len(threads)!=1:
        raise ValueError('Cannot identify one Desktop UI thread')
    thread = next(iter(threads))
    class Packet(ctypes.Structure):
        _fields_=[('state',w.LONG),('hr',w.LONG),('thread',w.DWORD),('reserved',w.DWORD),
                  ('script',w.WCHAR*8192),('result',w.WCHAR*32768)]
    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),
    }
    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
    # PID discovery records compatibility and requires a unique active UI host.
    if pid != detect_desktop_pid():
        raise ValueError('Explicit PID is not the validated active Desktop instance')
    dll=build_native_probe()
    module=hook=mapping=view=None
    report={'pid':pid,'pid_auto_detected':automatic_pid,'backend':'python-win32-native',
            'frida_used':False,'mode':config['mode'],'passed':False,'javascript_executed':False,
            'ui_thread_id':thread,'native_helper_path':str(dll)}
    report.update({'test_count':len(config.get('cases',[]))} if search_query is None else {'query':search_query,'limit':limit})
    try:
        mapping=kernel.CreateFileMappingW(w.HANDLE(-1),None,4,0,ctypes.sizeof(Packet),
                                         'Local\\WhatsAppLocalProbe-'+str(pid))
        mapping_error=ctypes.get_last_error()
        if not mapping:raise ctypes.WinError(mapping_error)
        if mapping_error==183:raise ValueError('Another native probe already owns this Desktop instance')
        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.state=3
        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())
        deadline=time.monotonic()+30
        def execute(expression):
            if len(expression.encode('utf-16-le'))//2>=8192:raise ValueError('Probe expression too long')
            packet.script=expression;packet.result='';packet.hr=0;packet.state=0
            while time.monotonic()<deadline:
                if not user.PostThreadMessageW(thread,0x8000+0x497,0,0):
                    raise ctypes.WinError(ctypes.get_last_error())
                time.sleep(.025)
                if packet.state==1:report['javascript_executed']=True
                if packet.state==-1:raise ValueError('Native HRESULT 0x'+format(packet.hr&0xffffffff,'08x'))
                if packet.state==2:
                    report['javascript_executed']=True
                    return packet.result
            if packet.state==0:
                raise TimeoutError('Native hook did not acknowledge the request; check helper path accessibility in the packaged Desktop process')
            raise TimeoutError('Timed out waiting for native WebView2 completion (state='+str(packet.state)+')')
        result=execute(config['start'])
        if config['mode']=='arithmetic':
            if config.get('cases'):
                import math
                parsed=json.loads(result)
                actual=parsed.get('cases',[])
                checks=[]
                if parsed.get('probe_id')!=config['probe_id'] or len(actual)!=len(config['cases']):
                    raise ValueError('Arithmetic result does not match this test session')
                for expected,observed in zip(config['cases'],actual):
                    value=observed.get('value')
                    target=expected['expected']
                    same=observed.get('expression')==expected['expression']
                    if isinstance(target,float):
                        same=same and isinstance(value,(int,float)) and not isinstance(value,bool) and math.isclose(value,target,rel_tol=1e-12,abs_tol=1e-12)
                    else:
                        same=same and type(value) is type(target) and value==target
                    checks.append({'expression':expected['expression'],'expected':target,'actual':value,'passed':same})
                report.update({'probe_id':config['probe_id'],'tests':checks,'passed':all(c['passed'] for c in checks)})
            else:
                report.update({'result':result,'passed':result=='2'})
        else:
            parsed=json.loads(result)
            deadline = time.monotonic() + 30
            while parsed.get('state') in ('started','pending'):
                if time.monotonic() >= deadline:
                    raise TimeoutError('Desktop search did not finish within 30 seconds')
                time.sleep(.2);parsed=json.loads(execute(config['poll']))
            report.update({'search':parsed,'passed':parsed.get('state')=='done'})
    except Exception as error:
        report['error']=str(error)
    finally:
        if hook and view:
            packet.state=6
            close_deadline=time.monotonic()+3
            while packet.state!=7 and time.monotonic()<close_deadline:
                user.PostThreadMessageW(thread,0x8000+0x497,0,0);time.sleep(.025)
            if packet.state!=7:report['cleanup_error']='Owner-thread cleanup did not acknowledge completion'
        if hook:user.UnhookWindowsHookEx(hook)
        if module:kernel.FreeLibrary(module)
        if view:kernel.UnmapViewOfFile(view)
        if mapping:kernel.CloseHandle(mapping)
        output.parent.mkdir(parents=True,exist_ok=True)
        output.write_text(json.dumps(report,indent=2)+'\n',encoding='utf-8')
    print(json.dumps(report,indent=2))
    if report.get('cleanup_error'):
        report['passed'] = False
        output.write_text(json.dumps(report,indent=2)+'\n',encoding='utf-8')
    return 0 if report.get('passed') else 1


def run_live_probe(pid, output, search_query=None, limit=50):
    """Run the bounded native WebView2 probe without Frida."""
    import uuid
    if search_query is None:
        raise ValueError('A search query is required; use the live math suite for arithmetic')
    if not search_query.strip() or len(search_query) > 200 or not 1 <= limit <= 50:
        raise ValueError('Search requires 1..200 characters and a limit of 1..50')
    key = '__local_search_probe_' + uuid.uuid4().hex
    # JSON serialization keeps the keyword a JavaScript string literal.
    query_literal, key_literal = json.dumps(search_query), json.dumps(key)
    start = r'''(() => {
      const key=__KEY__,query=__QUERY__,limit=__LIMIT__;
      if(typeof require!=='function')return {state:'error',error:'WhatsApp module loader unavailable'};
      window[key]={state:'pending'};
      Promise.resolve().then(()=>{
        const collection=require('WAWebMsgCollection').MsgCollection;
        if(!collection||typeof collection.search!=='function')throw Error('MsgCollection.search unavailable');
        return collection.search(query,1,limit);
      }).then(result=>{
        const messages=Array.isArray(result.messages)?result.messages:[];
        window[key]={state:'done',query,entry:'WAWebMsgCollection.MsgCollection.search',
          returned:messages.length,eof:result.eof,canceled:result.canceled,
          messages:messages.slice(0,limit).map(message=>{
            const body=typeof message.body==='string'?message.body:'';
            const caption=typeof message.caption==='string'?message.caption:'';
            return {id:message.id?.toString()??null,chat_id:message.id?.remote?.toString()??null,
              timestamp:message.t??null,type:message.type??null,
              body_length:body.length,caption_length:caption.length,
              contains_query:(body+'\n'+caption).toLowerCase().includes(query.toLowerCase())};
          })};
      }).catch(error=>{window[key]={state:'error',error:String(error),query};});
      return {state:'started',query};
    })()'''.replace('__KEY__',key_literal).replace('__QUERY__',query_literal).replace('__LIMIT__',str(limit))
    poll = r'''(() => {
      const key=__KEY__,value=window[key];
      if(!value)return {state:'error',error:'Search state missing'};
      if(value.state==='done'||value.state==='error')delete window[key];
      return value;
    })()'''.replace('__KEY__',key_literal)
    config = {'mode': 'search', 'start': start, 'poll': poll}
    return native_webview_test(pid, output, config, search_query, limit)