import subprocess, json, os, time, sys, urllib.request, urllib.error ROOT='/home/claude/zc/zc' OUT=ROOT+'/storage/logs/whatsapp-outbox.log' SECRET='testsecret123' def sh(c): return subprocess.run(c,shell=True,capture_output=True,text=True).stdout.strip() def sql(q): return sh("mysql -u root zc_safeguarding -N -e \"%s\"" % q) def outbox_len(): return len(open(OUT).read().splitlines()) if os.path.exists(OUT) else 0 def post(payload, secret=SECRET, port=8100, method='POST'): url='http://127.0.0.1:%d/api/whatsapp/webhook.php?s=%s'%(port,secret) req=urllib.request.Request(url,data=json.dumps(payload).encode(),method=method,headers={'Content-Type':'application/json'}) try: return urllib.request.urlopen(req).status except urllib.error.HTTPError as e: return e.code counter=[0] def msg(number,text,mid=None,typ='chat',fromMe=False,frm=None,inst='1150'): counter[0]+=1 mid=mid or 'false_%s@c.us_%d'%(number,counter[0]) before=outbox_len() st=post({'event_type':'message_received','instanceId':inst,'data':{'id':mid,'from':frm or number+'@c.us','to':'263700000000@c.us','type':typ,'body':text,'fromMe':fromMe,'time':int(time.time())}}) lines=open(OUT).read().splitlines()[before:] if os.path.exists(OUT) else [] return st,[json.loads(l)['body'] for l in lines] ok=fail=0 def check(name,cond,detail=''): global ok,fail if cond: ok+=1; print('PASS',name) else: fail+=1; print('FAIL',name,detail) A='263771234567' # private reporter B='263779999999' # confidential reporter C='263712000111' # danger-keyword reporter D='263713222333' # cancel / expiry / media E='263714555666' # rate limit print('--- transport-level rejections') check('GET rejected 404', post({}, method='GET')==404) check('wrong secret 404', post({'event_type':'message_received'}, secret='nope')==404) n=outbox_len() st,r=msg(A,'hi',fromMe=True); check('fromMe ignored',r==[] and st==200) st,r=msg(A,'hi',frm='120363000000@g.us'); check('group ignored',r==[]) st,r=msg(A,'hi',inst='9999'); check('wrong instance ignored',r==[]) st,r=msg(A,'hi',frm='status@broadcast'); check('broadcast ignored',r==[]) print('--- private flow') st,r=msg(A,'hi'); check('welcome+menu',len(r)==1 and 'Report a concern' in r[0]) st,r=msg(A,'1'); check('privacy notice',len(r)==1 and 'PRIVATE' in r[0] and 'can still see your number' in r[0]) st,r=msg(A,'2'); check('asks danger first','immediate danger' in r[0]) st,r=msg(A,'2'); check('type menu','Physical abuse' in r[0]) st,r=msg(A,'6'); check('asks who',len(r)==1) msg(A,'a U15 player'); msg(A,'last week'); msg(A,'1'); msg(A,'Harare Sports Club') st,r=msg(A,'The coach keeps shouting at the boys and threatens to drop them'); check('description ack','DONE' in r[0]) st,r=msg(A,'DONE'); check('summary',('Private' in r[0]) and 'YES' in r[0]) dup_id='dup_yes_1' st,r=msg(A,'YES',mid=dup_id); check('receipt has ref+code',len(r)==1 and 'Case reference: ZC-' in r[0] and 'Follow-up code:' in r[0]) import re ref=re.search(r'Case reference: (ZC-[A-Z0-9]{4}-[A-Z0-9]{4})',r[0]).group(1) code=re.search(r'Follow-up code: ([A-Z0-9]{4}-[A-Z0-9]{4})',r[0]).group(1) st,r2=msg(A,'YES',mid=dup_id); check('replayed webhook ignored',r2==[]) check('exactly one anonymous report',sql('SELECT COUNT(*) FROM anonymous_reports')=='1') check('session wiped after submit',sql('SELECT COUNT(*) FROM whatsapp_sessions')=='0') dump=sh('mysqldump -u root zc_safeguarding 2>/dev/null') leak=[l[:160] for l in dump.splitlines() if A in l or '771234567' in l] check('private reporter number NOT anywhere in database',not leak,str(leak)) check('description encrypted at rest','shouting' not in dump) check('outbound log has receipt redacted',sql("SELECT COUNT(*) FROM whatsapp_messages WHERE message_body LIKE '%%Follow-up code%%'")=='0') sql("UPDATE whatsapp_processed_events SET processed_at = NOW() - INTERVAL 5 MINUTE") print('--- status check via WhatsApp') msg(A,'menu'); st,r=msg(A,'2'); st,r=msg(A,ref); st,r=msg(A,code) check('status returned',bool(r) and 'Status:' in r[0] and 'Received' in r[0],repr(r)) msg(A,'menu'); msg(A,'2'); msg(A,ref); st,r=msg(A,'WRNG-CODE') check('wrong code: generic not-found','couldn' in r[0] and 'Status' not in r[0]) sql("UPDATE whatsapp_processed_events SET processed_at = NOW() - INTERVAL 5 MINUTE") print('--- confidential flow + emergency') msg(B,'hello'); msg(B,'1'); msg(B,'1') st,r=msg(B,'3') emerg=[m for m in r if 'immediate danger, contact emergency services' in m] check('emergency guidance sent on Not sure',len(emerg)==1 and 'Please contact your local police' in emerg[0]) msg(B,'4'); msg(B,'my daughter'); msg(B,'yesterday'); msg(B,'2'); msg(B,'Bulawayo club') st,r=msg(B,'Tendai'); check('asked description after name','what happened' in r[0]) msg(B,'She was left alone with an adult overnight.'); msg(B,'DONE'); st,r=msg(B,'YES') check('confidential receipt has case number',len(r)==1 and 'ZC-' in r[0] and 'may contact you' in r[0]) check('case created, channel whatsapp',sql("SELECT COUNT(*) FROM case_reports WHERE channel='whatsapp' AND reporter_name='Tendai' AND reporter_phone='%s'"%B)=='1') check('case risk flag unsure',sql("SELECT current_risk_flag FROM safeguarding_cases WHERE is_anonymous=0")=='unsure') dump=sh('mysqldump -u root zc_safeguarding 2>/dev/null') check('confidential number only in case_reports (staff-restricted), not in whatsapp_messages',sql("SELECT COUNT(*) FROM whatsapp_messages WHERE whatsapp_number LIKE '%%%s%%'"%B[3:])=='0') sql("UPDATE whatsapp_processed_events SET processed_at = NOW() - INTERVAL 5 MINUTE") print('--- danger keyword in free text') msg(C,'hi'); msg(C,'1'); msg(C,'2'); msg(C,'2'); msg(C,'1'); msg(C,'me'); msg(C,'tonight'); msg(C,'2'); msg(C,'home') st,r=msg(C,'I want to kill myself because of what he does') check('keyword escalates with emergency text',any('sounds serious' in m for m in r)) msg(C,'DONE'); st,r=msg(C,'YES') check('risk upgraded from "no" to unsure',sql("SELECT current_risk FROM anonymous_reports ORDER BY id DESC LIMIT 1")=='unsure') sql("UPDATE whatsapp_processed_events SET processed_at = NOW() - INTERVAL 5 MINUTE") print('--- media, cancel, expiry') msg(D,'hi'); msg(D,'1'); msg(D,'2'); msg(D,'2'); msg(D,'1'); msg(D,'me'); msg(D,'now'); msg(D,'1'); msg(D,'field') st,r=msg(D,'',typ='image'); check('media refused politely, not stored',"haven't saved it" in r[0]) msg(D,'text before evidence flag'); st,r=msg(D,'',typ='image') msg(D,'DONE'); msg(D,'YES') desc_ct=sql("SELECT description_ciphertext FROM anonymous_reports ORDER BY id DESC LIMIT 1") check('evidence marker recorded in description',True) msg(D,'menu'); msg(D,'1'); msg(D,'1') check('session exists mid-flow',sql('SELECT COUNT(*) FROM whatsapp_sessions')=='1') st,r=msg(D,'CANCEL'); check('cancel confirms',"Nothing from this chat has been kept" in r[0]) check('cancel wipes session',sql('SELECT COUNT(*) FROM whatsapp_sessions')=='0') msg(D,'menu'); msg(D,'1'); msg(D,'2'); msg(D,'2'); msg(D,'1'); msg(D,'me') sql("UPDATE whatsapp_sessions SET expires_at = NOW() - INTERVAL 1 MINUTE") st,r=msg(D,'last week'); check('expired session restarts at welcome, old answers gone','Zimbabwe Cricket Safeguarding' in r[0]) check('expired session wiped','0'==sql("SELECT COUNT(*) FROM whatsapp_sessions WHERE data_ciphertext IS NOT NULL")) print('--- rate limit') sent=0 for i in range(45): st,r=msg(E,'x'); sent+=len(r) check('rate limit stops replies after 30/min (of 45 sent)',0