129 lines
9.4 KiB
Python
129 lines
9.4 KiB
Python
"""Request a real WhatsApp mobile-linking QR in an isolated background browser.
|
|
|
|
The official WhatsApp Web client creates the keys and requests the server ref.
|
|
This script reads the displayed QR payload and renders it locally. It does not
|
|
extract credentials from the existing Desktop account or implement fake pairing.
|
|
Requires: pip install websockets 'qrcode[pil]'
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse, base64, datetime as dt, hashlib, json, os
|
|
from pathlib import Path
|
|
import socket, subprocess, sys, time, urllib.parse, urllib.request
|
|
import http.server,threading
|
|
|
|
from .common import BASE as ROOT
|
|
READ_QR=r'''(() => {
|
|
const element=Array.from(document.querySelectorAll('[data-ref]')).find(e=>{const v=e.getAttribute('data-ref');return v&&v.split(',').length>=4;});
|
|
const payload=element?.getAttribute('data-ref')??null;
|
|
let linked=false;
|
|
try {linked=!!require('WAWebUserPrefsMeUser').getMaybeMePnUser();}catch{}
|
|
let refreshRequested=false;
|
|
if(!linked&&!payload){const button=document.querySelector('[data-testid="link_device_qr_expired_refresh_button"]');if(button){button.click();refreshRequested=true;}}
|
|
return {payload,linked,refreshRequested,title:document.title,origin:location.origin};
|
|
})()'''
|
|
|
|
def chrome_path(explicit):
|
|
if explicit:return Path(explicit).resolve()
|
|
candidates=[Path(os.environ.get('PROGRAMFILES','C:/Program Files'))/'Google/Chrome/Application/chrome.exe',Path(os.environ.get('PROGRAMFILES(X86)','C:/Program Files (x86)'))/'Google/Chrome/Application/chrome.exe',Path(os.environ.get('LOCALAPPDATA',''))/'Google/Chrome/Application/chrome.exe']
|
|
return next((p for p in candidates if p.is_file()),None)
|
|
|
|
def call(url,expression):
|
|
from websockets.sync.client import connect
|
|
with connect(url,open_timeout=10,max_size=1_000_000) as ws:
|
|
ws.send(json.dumps({'id':1,'method':'Runtime.evaluate','params':{'expression':expression,'returnByValue':True}}))
|
|
while True:
|
|
result=json.loads(ws.recv(timeout=15))
|
|
if result.get('id')!=1:continue
|
|
if result.get('error') or result.get('result',{}).get('exceptionDetails'):raise RuntimeError('Page probe failed')
|
|
return result['result']['result'].get('value')
|
|
|
|
def describe(payload):
|
|
parts=payload.split(',')
|
|
lengths=[]
|
|
for part in parts[1:]:
|
|
try:lengths.append(len(base64.b64decode(part,validate=True)))
|
|
except Exception:lengths.append(None)
|
|
return {'field_count':len(parts),'payload_length':len(payload),'sha256':hashlib.sha256(payload.encode()).hexdigest(),'base64_field_byte_lengths':lengths}
|
|
|
|
def write_qr(payload,out):
|
|
import qrcode
|
|
qr=qrcode.QRCode(error_correction=qrcode.constants.ERROR_CORRECT_M,box_size=10,border=4)
|
|
qr.add_data(payload);qr.make(fit=True)
|
|
qr.make_image(fill_color='black',back_color='white').save(out/'pairing-qr.png')
|
|
# No raw QR payload, identity secret or session key is written to the report.
|
|
report={'timestamp_utc':dt.datetime.now(dt.timezone.utc).isoformat(),'status':'waiting_for_mobile_scan','source':'Official WhatsApp Web QR DOM in isolated headless Chrome','qr_metadata':describe(payload),'note':'QR rotates and requires the matching browser session to remain running.'}
|
|
(out/'pairing-status.json').write_text(json.dumps(report,indent=2),encoding='utf-8')
|
|
(out/'pairing-qr.html').write_text('<!doctype html><html><head><meta charset="utf-8"><meta http-equiv="refresh" content="3"><title>WhatsApp test pairing</title></head><body style="font-family:Arial;text-align:center"><h1>Link your own WhatsApp phone</h1><p>WhatsApp → Linked devices → Link a device.</p><img src="pairing-qr.png" style="max-width:90vw"><p>This QR links the isolated Python test session. The image updates while the script runs.</p></body></html>',encoding='utf-8')
|
|
|
|
def serve_viewer(out):
|
|
class Handler(http.server.SimpleHTTPRequestHandler):
|
|
def __init__(self,*a,**kw):super().__init__(*a,directory=str(out),**kw)
|
|
def log_message(self,*a):pass
|
|
def do_GET(self):
|
|
if urllib.parse.urlsplit(self.path).path not in ('/pairing-qr.html','/pairing-qr.png','/pairing-status.json'):
|
|
self.send_error(404);return
|
|
super().do_GET()
|
|
server=http.server.ThreadingHTTPServer(('127.0.0.1',0),Handler)
|
|
threading.Thread(target=server.serve_forever,daemon=True).start()
|
|
return server,f'http://127.0.0.1:{server.server_port}/pairing-qr.html'
|
|
|
|
def main(argv=None):
|
|
p=argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--browser',type=Path);p.add_argument('--profile',type=Path,default=ROOT/'whatsapp-pairing-profile');p.add_argument('--output-dir',type=Path,default=ROOT/'whatsapp-pairing');p.add_argument('--watch-seconds',type=int,default=300);p.add_argument('--startup-seconds',type=int,default=90)
|
|
args=p.parse_args(argv);browser=chrome_path(args.browser)
|
|
if not browser or not browser.is_file():p.error('Chrome executable missing; supply --browser')
|
|
if not 10<=args.watch_seconds<=3600:p.error('--watch-seconds must be 10..3600')
|
|
profile=args.profile.resolve();out=args.output_dir.resolve()
|
|
if '5319275a.whatsappdesktop' in str(profile).lower():p.error('Refusing the existing WhatsApp Desktop profile')
|
|
marker=profile/'python-pairing-profile.json'
|
|
if profile.exists() and any(profile.iterdir()) and not marker.exists():p.error('Profile must be empty or created by this script')
|
|
profile.mkdir(parents=True,exist_ok=True);out.mkdir(parents=True,exist_ok=True)
|
|
marker.write_text(json.dumps({'purpose':'isolated official WhatsApp Web pairing test'}),encoding='utf-8')
|
|
with socket.socket() as sock:sock.bind(('127.0.0.1',0));port=sock.getsockname()[1]
|
|
command=[str(browser),'--headless=new','--no-first-run','--no-default-browser-check','--remote-debugging-address=127.0.0.1',f'--remote-debugging-port={port}',f'--user-data-dir={profile}','--window-size=1200,1000','https://web.whatsapp.com/']
|
|
flags=subprocess.CREATE_NO_WINDOW if sys.platform=='win32' else 0
|
|
proc=subprocess.Popen(command,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,creationflags=flags)
|
|
viewer,viewer_url=serve_viewer(out)
|
|
endpoint=f'http://127.0.0.1:{port}';(out/'pairing-runtime.json').write_text(json.dumps({'browser_pid':proc.pid,'cdp_endpoint':endpoint,'viewer_url':viewer_url,'profile':str(profile)},indent=2),encoding='utf-8')
|
|
start=time.monotonic();last_payload=None;generated=0;linked=False
|
|
print(f'Isolated background session started; QR output: {out}',flush=True)
|
|
print(f'REA observation endpoint: {endpoint}',flush=True)
|
|
print(f'QR viewer (auto-refresh): {viewer_url}',flush=True)
|
|
try:
|
|
while time.monotonic()-start<args.watch_seconds:
|
|
if proc.poll() is not None:raise RuntimeError('Background browser exited')
|
|
try:
|
|
with urllib.request.urlopen(endpoint+'/json/list',timeout=3) as response:targets=json.load(response)
|
|
target=next((t for t in targets if t.get('type')=='page' and urllib.parse.urlsplit(t.get('url','')).hostname=='web.whatsapp.com'),None)
|
|
if not target:time.sleep(1);continue
|
|
state=call(target['webSocketDebuggerUrl'],READ_QR)
|
|
except (OSError,StopIteration):time.sleep(1);continue
|
|
if not state or state.get('origin')!='https://web.whatsapp.com':
|
|
if time.monotonic()-start>args.startup_seconds:raise RuntimeError('WhatsApp page did not commit before startup timeout')
|
|
time.sleep(1);continue
|
|
if state.get('linked'):
|
|
linked=True;(out/'pairing-status.json').write_text(json.dumps({'status':'linked','timestamp_utc':dt.datetime.now(dt.timezone.utc).isoformat(),'profile':str(profile),'note':'Browser is kept alive until the watch timeout; credentials remain only in its isolated profile.'},indent=2),encoding='utf-8')
|
|
if last_payload is not None:print('Mobile linking confirmed.',flush=True);last_payload=None
|
|
elif state.get('payload') and state['payload']!=last_payload:
|
|
last_payload=state['payload'];write_qr(last_payload,out);generated+=1
|
|
print(f'QR refreshed ({generated}); scan pairing-qr.png from your own phone.',flush=True)
|
|
elif state.get('refreshRequested'):
|
|
print('Expired QR: requested a fresh reference through the official reload button.',flush=True)
|
|
elif not generated and time.monotonic()-start>args.startup_seconds:raise RuntimeError('No pairing QR appeared before startup timeout')
|
|
time.sleep(2)
|
|
return 0 if generated or linked else 1
|
|
except KeyboardInterrupt:return 0
|
|
finally:
|
|
viewer.shutdown();viewer.server_close()
|
|
try:
|
|
with urllib.request.urlopen(endpoint+'/json/version',timeout=3) as response:version=json.load(response)
|
|
with connect(version['webSocketDebuggerUrl'],open_timeout=3) as ws:ws.send(json.dumps({'id':1,'method':'Browser.close'}))
|
|
except Exception:proc.terminate()
|
|
try:proc.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:proc.kill()
|
|
(out/'pairing-runtime.json').write_text(json.dumps({'running':False,'linked':linked,'qr_refreshes':generated},indent=2),encoding='utf-8')
|
|
if not linked and (out/'pairing-status.json').exists():
|
|
report=json.loads((out/'pairing-status.json').read_text());report['status']='expired_browser_closed';(out/'pairing-status.json').write_text(json.dumps(report,indent=2),encoding='utf-8')
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|