karl/ghostwire
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ghostwire/whatsapp_suite/webhook.py
T

481 lines
24 KiB
Python
Raw Permalink Normal View History

Add autonomous webhooks and dynamic WebView discovery
2026-10-11 08:43:39 +02:00
"""Autonomous in-process message listener with HTTPS or loopback delivery.
The install command exits; a resident native timer relays queued JavaScript
events. The optional receiver restore loop reattaches after app/page restarts.
"""
import argparse
import base64
import os
import re
from urllib.parse import urlsplit
from contextlib import closing
import ctypes
from ctypes import wintypes as w
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from .common import BASE, asset_bytes, verified_helper, cli_command
from .desktop import detect_desktop_pid, visible_ui_threads
from .live import Packet, run_math, windows_api, wait_for_state
from .zip_export import evaluate
HELPER_SHA256 = '10acf825aa3c37f356bff6639534db192a5e9babb0402e5d0651f8aecf8c2101'
WAKE = 0x8000 + 0x499
PROTOCOL = 0x57484231
STATE_KEY = '__whatsappWebhookV1'
def validate_config(config):
if not isinstance(config, dict) or config.get('version') != 1:
raise ValueError('Unsupported webhook configuration')
if config.get('endpoint'):
endpoint=urlsplit(config['endpoint'])
if (endpoint.scheme!='https' or not endpoint.hostname or endpoint.username or endpoint.password
or endpoint.fragment or endpoint.query or not endpoint.path.startswith('/')
or '\n' in config['endpoint'] or '\r' in config['endpoint']):
raise ValueError('Direct endpoint must be an HTTPS URL without credentials, query or fragment')
if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*',config.get('token_env','')):
raise ValueError('token_env must name a credential environment variable')
elif type(config.get('port')) is not int or not 1024 <= config['port'] <= 65535:
raise ValueError('Loopback receiver port must be between 1024 and 65535')
if type(config.get('self_only')) is not bool:
raise ValueError('self_only must be a boolean')
if type(config.get('enabled', True)) is not bool:
raise ValueError('enabled must be a boolean')
if type(config.get('since_ms')) is not int or config['since_ms'] < 0:
raise ValueError('Invalid listener start time')
uuid.UUID(config.get('listener_id', ''))
return config
def read_config(path):
return validate_config(json.loads(Path(path).read_text(encoding='utf-8')))
def save_config(path, config):
"""Replace complete configuration so a restore check never sees partial JSON."""
path = Path(path)
descriptor, temporary = tempfile.mkstemp(prefix='listener-', suffix='.tmp', dir=path.parent)
try:
with os.fdopen(descriptor, 'w', encoding='utf-8') as stream:
json.dump(config, stream, indent=2)
os.replace(temporary, path)
finally:
Path(temporary).unlink(missing_ok=True)
def protect_token(value, decrypt=False):
"""Windows current-user DPAPI; persisted configuration contains no raw token."""
if os.name != 'nt':
raise ValueError('Token protection requires Windows DPAPI')
class Blob(ctypes.Structure):
_fields_ = [('length', w.DWORD), ('data', ctypes.POINTER(ctypes.c_ubyte))]
raw = base64.b64decode(value, validate=True) if decrypt else value.encode('ascii')
buffer = ctypes.create_string_buffer(raw)
source = Blob(len(raw), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_ubyte)))
output = Blob()
crypt = ctypes.WinDLL('crypt32', use_last_error=True)
kernel = ctypes.WinDLL('kernel32', use_last_error=True)
kernel.LocalFree.argtypes = [ctypes.c_void_p]
kernel.LocalFree.restype = ctypes.c_void_p
if decrypt:
call = crypt.CryptUnprotectData
call.argtypes = [ctypes.POINTER(Blob), ctypes.c_void_p, ctypes.c_void_p,
ctypes.c_void_p, ctypes.c_void_p, w.DWORD, ctypes.POINTER(Blob)]
arguments = (ctypes.byref(source), None, None, None, None, 1, ctypes.byref(output))
else:
call = crypt.CryptProtectData
call.argtypes = [ctypes.POINTER(Blob), w.LPCWSTR, ctypes.c_void_p,
ctypes.c_void_p, ctypes.c_void_p, w.DWORD, ctypes.POINTER(Blob)]
arguments = (ctypes.byref(source), 'WhatsApp injector webhook token', None,
None, None, 1, ctypes.byref(output))
call.restype = w.BOOL
try:
if not call(*arguments):
raise ctypes.WinError(ctypes.get_last_error())
data = ctypes.string_at(output.data, output.length)
return data.decode('ascii') if decrypt else base64.b64encode(data).decode('ascii')
finally:
ctypes.memset(buffer, 0, len(raw))
if output.data:
ctypes.memset(output.data, 0, output.length)
kernel.LocalFree(output.data)
def configuration_token(config):
value = os.environ.get(config.get('token_env', ''))
if not value and config.get('protected_token'):
value = protect_token(config['protected_token'], decrypt=True)
return value
def register_restore_task(config_path):
"""One per-user scheduled check, not a resident Python service."""
config_path = Path(config_path).resolve()
config = read_config(config_path)
if not config.get('endpoint') or os.name != 'nt':
raise ValueError('Scheduled restoration requires a direct HTTPS endpoint on Windows')
token = configuration_token(config)
if not token or not re.fullmatch(r'[!-~]{32,128}', token):
raise ValueError('A valid webhook token is required for automatic restoration')
config['protected_token'] = protect_token(token)
name = 'WhatsAppInjector-' + config['listener_id']
command = cli_command('webhook', 'ensure', '--config', str(config_path))
if command[1] == '-m':
bundle = BASE / 'whatsapp.py'
if not bundle.exists():
raise ValueError('Build whatsapp.py before registering a restore task')
command = [sys.executable, str(bundle), 'webhook', 'ensure', '--config', str(config_path)]
executable = Path(command[0])
windowless = executable.with_name('pythonw.exe')
if windowless.exists():
executable = windowless
arguments = subprocess.list2cmdline(command[1:])
quote = lambda value: "'" + str(value).replace("'", "''") + "'"
script = '''$ErrorActionPreference='Stop'
$taskName=__NAME__
$taskAction=New-ScheduledTaskAction -Execute __EXE__ -Argument __ARGS__ -WorkingDirectory __CWD__
$taskTrigger=New-ScheduledTaskTrigger -Once -At (Get-Date).AddMinutes(1) -RepetitionInterval ([TimeSpan]::FromMinutes(1))
$taskPrincipal=New-ScheduledTaskPrincipal -UserId ([System.Security.Principal.WindowsIdentity]::GetCurrent().Name) -LogonType Interactive -RunLevel Limited
$taskSettings=New-ScheduledTaskSettingsSet -Hidden -StartWhenAvailable -MultipleInstances IgnoreNew -ExecutionTimeLimit ([TimeSpan]::FromSeconds(30)) -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries
$existing=Get-ScheduledTask -TaskName $taskName -ErrorAction SilentlyContinue
if($existing){
if($existing.Actions.Execute -ne $taskAction.Execute -or $existing.Actions.Arguments -ne $taskAction.Arguments){throw 'Task name belongs to another action'}
Enable-ScheduledTask -TaskName $taskName | Out-Null
}else{
Register-ScheduledTask -TaskName $taskName -Action $taskAction -Trigger $taskTrigger -Principal $taskPrincipal -Settings $taskSettings -Description 'Restore explicitly configured WhatsApp message webhook when the app is running' | Out-Null
}
'''
for marker, value in [('__NAME__', name), ('__EXE__', executable),
('__ARGS__', arguments), ('__CWD__', config_path.parent)]:
script = script.replace(marker, quote(value))
# Protected token is saved before the task can run; it is never a task argument.
save_config(config_path, config)
result = subprocess.run(['powershell.exe', '-NoProfile', '-NonInteractive', '-Command', script],
capture_output=True, text=True, timeout=30)
if result.returncode:
raise RuntimeError('Could not register the per-user restore task: ' + result.stderr[:500])
config['restore_task'] = name
save_config(config_path, config)
return {'task': name, 'interval_seconds': 60, 'token_protection': 'Windows current-user DPAPI'}
def ensure_listener(config_path):
config = read_config(config_path)
if not config.get('enabled', True):
return {'passed': True, 'disabled': True}
report = manage('status')
if not report['passed']:
return {'passed': False, 'error': report.get('error'), 'waiting_for_whatsapp': True}
listener, relay = report.get('listener', {}), report.get('relay', {})
if listener.get('installed') and listener.get('listener_id') != config['listener_id']:
raise ValueError('Another listener is active; refusing replacement')
if not listener.get('installed') or not relay.get('running'):
report = manage('install', config)
report['restored'] = report['passed']
return report
def native_control(pid, command, port=0, endpoint=None, token=None):
"""Install/status/stop using a short-lived hook; relay holds its own lifetime."""
kernel, user = windows_api()
threads = visible_ui_threads(pid)
if len(threads) != 1:
raise ValueError('Expected one WhatsApp UI thread')
thread = next(iter(threads))
dll, _ = verified_helper('webhook_probe.dll', HELPER_SHA256,
'.whatsapp-webhook-native', 'webhook_probe.dll')
module = hook = mapping = view = process = None
try:
process = kernel.OpenProcess(0x100000, False, pid)
if not process:
raise ctypes.WinError(ctypes.get_last_error())
ctypes.set_last_error(0)
mapping = kernel.CreateFileMappingW(w.HANDLE(-1), None, 4, 0,
ctypes.sizeof(Packet), 'Local\\WhatsAppWebhook-v1-' + str(pid))
existed = ctypes.get_last_error() == 183
if not mapping:
raise ctypes.WinError(ctypes.get_last_error())
if existed:
raise ValueError('Another webhook controller is using this 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, packet.protocol = thread, PROTOCOL
packet.state = {'install': 0, 'status': 3, 'stop': 6}[command]
if command=='install' and endpoint:
if not token or not re.fullmatch(r'[!-~]{32,128}',token):
raise ValueError('Webhook token must contain 32-128 printable ASCII characters')
packet.script=endpoint+'\n'+token
else:
packet.script = str(port) if command == 'install' else ''
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
wait_for_state(packet, {'install': 2, 'status': 4, 'stop': 7}[command],
10, wake, alive, 'webhook ' + command)
return json.loads(packet.result)
finally:
unhook_error = None
if hook and not user.UnhookWindowsHookEx(hook):
unhook_error = ctypes.WinError(ctypes.get_last_error())
if view:
kernel.UnmapViewOfFile(view)
for handle in (mapping, process):
if handle:
kernel.CloseHandle(handle)
if module:
kernel.FreeLibrary(module)
if unhook_error:
raise unhook_error
def manage(action, config=None, pid=None):
if action == 'install':
validate_config(config)
def operation(execute, session, report):
if action == 'install':
source = asset_bytes('webhook_listener.js').decode('utf-8')
settings = {key: config[key] for key in ('listener_id', 'self_only', 'since_ms')}
settings['reconcile'] = True
source = source.replace('__CONFIG__', json.dumps(settings))
report['listener'] = evaluate(execute, source, session)
try:
report['relay'] = native_control(report['pid'], 'install', config.get('port',0),
config.get('endpoint'),configuration_token(config) if config.get('endpoint') else None)
except Exception:
evaluate(execute, 'window.' + STATE_KEY + '?.stop()', session)
raise
elif action == 'stop':
report['relay'] = native_control(report['pid'], 'stop')
report['listener'] = evaluate(execute,
'window.' + STATE_KEY + '?.stop() ?? {installed:false}', session)
else:
report['listener'] = evaluate(execute,
'window.' + STATE_KEY + '?.status() ?? {installed:false}', session)
report['relay'] = native_control(report['pid'], 'status')
return run_math(pid, 'webhook', _operation=operation)
def initialize_store(path):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with closing(sqlite3.connect(path)) as db, db:
db.execute('CREATE TABLE IF NOT EXISTS events '
'(event_id TEXT PRIMARY KEY, listener_id TEXT NOT NULL, '
'received_at REAL NOT NULL, payload TEXT NOT NULL)')
def store_events(path, payload, listener_id=None):
"""Commit before HTTP acknowledgement; duplicate deliveries are harmless."""
if not isinstance(payload, dict) or not isinstance(payload.get('events'), list):
raise ValueError('Expected webhook event batch')
if len(payload['events']) > 16:
raise ValueError('Too many events')
actual = payload.get('listener_id')
uuid.UUID(actual or '')
if listener_id is not None and actual != listener_id:
raise ValueError('Unexpected listener ID')
records = []
for event in payload['events']:
if (not isinstance(event, dict) or event.get('type') != 'message'
or type(event.get('seq')) is not int or event['seq'] <= 0
or not isinstance(event.get('event_id'), str)
or not 1 <= len(event['event_id']) <= 1024
or not isinstance(event.get('message'), dict)
or event['message'].get('id') != event['event_id']):
raise ValueError('Invalid message event')
records.append((event['event_id'], actual, time.time(), json.dumps(event, ensure_ascii=False)))
with closing(sqlite3.connect(path)) as db, db:
before = db.total_changes
db.executemany('INSERT OR IGNORE INTO events VALUES (?,?,?,?)', records)
return db.total_changes - before
def make_server(port, store, listener_id=None, health=None):
initialize_store(store)
health = health if health is not None else {}
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def reply(self, code, value):
data = json.dumps(value).encode('utf-8')
self.send_response(code)
self.send_header('Content-Type', 'application/json')
self.send_header('Content-Length', str(len(data)))
self.end_headers()
self.wfile.write(data)
def do_GET(self):
if self.path != '/health':
self.reply(404, {'error': 'Not found'})
return
with closing(sqlite3.connect(store)) as db:
count = db.execute('SELECT COUNT(*) FROM events').fetchone()[0]
self.reply(200, dict(health, stored_events=count))
def do_POST(self):
if self.path != '/whatsapp-webhook':
self.reply(404, {'error': 'Not found'})
return
try:
length = int(self.headers.get('Content-Length', '0'))
if not 0 < length <= 262144:
self.reply(413, {'error': 'Invalid payload size'})
return
self.connection.settimeout(5)
data = self.rfile.read(length)
if len(data) != length:
raise ValueError('Incomplete payload')
payload = json.loads(data)
inserted = store_events(store, payload, listener_id)
self.reply(200, {'accepted': True, 'inserted': inserted})
except (ValueError, TypeError, OSError):
self.reply(400, {'error': 'Invalid event batch'})
except sqlite3.Error:
self.reply(503, {'error': 'Storage unavailable; retry'})
return ThreadingHTTPServer(('127.0.0.1', port), Handler)
def restore_loop(config_path, health, stopped):
"""Receiver-owned bootstrap, needed when process memory is recreated."""
while not stopped.is_set():
try:
config = read_config(config_path)
if not config.get('enabled', True):
health.update(connected=False, disabled=True, last_error=None)
stopped.wait(5)
continue
health['disabled'] = False
report = manage('status')
listener, relay = report.get('listener', {}), report.get('relay', {})
if not report['passed']:
raise RuntimeError(report.get('error', 'Status failed'))
if listener.get('installed') and listener.get('listener_id') != config['listener_id']:
raise RuntimeError('A different listener is active; refusing replacement')
if not listener.get('installed') or not relay.get('running'):
if not read_config(config_path).get('enabled', True):
continue
report = manage('install', config)
if not report['passed']:
raise RuntimeError(report.get('error', 'Listener restore failed'))
health['restores'] = health.get('restores', 0) + 1
health.update(connected=True, pid=report['pid'], last_error=None)
except (OSError, ValueError, RuntimeError) as error:
health.update(connected=False, last_error=str(error)[:500])
stopped.wait(5)
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest='action', required=True)
install = commands.add_parser('install', help='Install listener and resident relay; this process exits')
install.add_argument('--port', type=int, default=18763)
install.add_argument('--endpoint',help='Direct HTTPS webhook; no local receiver needed')
install.add_argument('--token-env',default='WHATSAPP_INJECTOR_WEBHOOK_TOKEN',help='Environment variable holding direct-endpoint token')
install.add_argument('--self-only', action='store_true', help='Capture only your self-chat')
install.add_argument('--config', type=Path, default=Path('analysis/webhooks/listener.json'))
install.add_argument('--pid', type=int)
install.add_argument('--restore', action='store_true', help='Reuse the saved configuration')
install.add_argument('--auto-restore', action='store_true', help='Register a per-user minute check; requires a direct HTTPS endpoint')
for action in ('ensure', 'auto-restore'):
sub = commands.add_parser(action)
sub.add_argument('--config', type=Path, required=True)
for action in ('status', 'stop'):
sub = commands.add_parser(action)
sub.add_argument('--pid', type=int)
if action == 'stop':
sub.add_argument('--config', type=Path,
help='Disable automatic restore in this saved configuration')
watch = commands.add_parser('watch',help='Optional automatic restoration after WhatsApp restarts; keeps this process running')
watch.add_argument('--config',type=Path,required=True)
serve = commands.add_parser('serve', help='Run loopback receiver, optionally restoring after WhatsApp restarts')
serve.add_argument('--port', type=int, default=18763)
serve.add_argument('--store', type=Path, default=Path('analysis/webhooks/events.sqlite3'))
serve.add_argument('--restore-config', type=Path)
args = parser.parse_args(argv)
if args.action == 'auto-restore':
print(json.dumps(register_restore_task(args.config)))
return 0
if args.action == 'ensure':
report = ensure_listener(args.config)
summary = {k: report.get(k) for k in ('passed', 'pid', 'restored', 'disabled', 'waiting_for_whatsapp', 'error')}
summary['checked_at'] = time.time()
Path(args.config).with_suffix('.status.json').write_text(json.dumps(summary, indent=2), encoding='utf-8')
print(json.dumps({k: report.get(k) for k in ('passed', 'pid', 'restored', 'disabled', 'error')}))
return 0 if report['passed'] or report.get('waiting_for_whatsapp') else 1
if args.action == 'watch':
read_config(args.config)
stopped=threading.Event();health={}
print('Watching WhatsApp restarts; Ctrl+C stops automatic restoration.',flush=True)
try:
restore_loop(args.config,health,stopped)
except KeyboardInterrupt:
stopped.set()
return 0
if args.action == 'serve':
config = read_config(args.restore_config) if args.restore_config else None
if config and config.get('endpoint'):
raise ValueError('Direct endpoints do not use serve; use webhook watch for optional restoration')
port = config['port'] if config else args.port
if not 1024 <= port <= 65535:
raise ValueError('Invalid loopback port')
health, stopped = {}, threading.Event()
server = make_server(port, args.store, config['listener_id'] if config else None, health)
if config:
threading.Thread(target=restore_loop, args=(args.restore_config, health, stopped), daemon=True).start()
print(json.dumps({'listening': 'http://127.0.0.1:' + str(port), 'store': str(args.store),
'automatic_restore': bool(config)}), flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
stopped.set()
server.server_close()
return 0
config = None
if args.action == 'install':
if args.restore:
config = read_config(args.config)
config['enabled'] = True
else:
if args.config.exists():
raise ValueError('Configuration exists; use --restore or a new --config')
config = validate_config({'version': 1, 'port': args.port, 'self_only': args.self_only, 'enabled': True,
'since_ms': int(time.time() * 1000), 'listener_id': str(uuid.uuid4()),
**({'endpoint':args.endpoint,'token_env':args.token_env} if args.endpoint else {})})
if args.action == 'stop' and args.config:
saved = read_config(args.config)
saved['enabled'] = False
save_config(args.config, saved)
report = manage(args.action, config, args.pid)
if args.action == 'install' and report['passed']:
args.config.parent.mkdir(parents=True, exist_ok=True)
with args.config.open('w' if args.restore else 'x', encoding='utf-8') as stream:
json.dump(config, stream, indent=2)
if args.auto_restore:
report['automatic_restore'] = register_restore_task(args.config)
print(json.dumps({k: report.get(k) for k in ('passed', 'pid', 'listener', 'relay', 'automatic_restore', 'error')}))
return 0 if report['passed'] else 1
Reference in new issue Copy Permalink