import json, urllib.request, sqlite3, uuid, io, struct, zlib B='http://127.0.0.1:8090/api' db=sqlite3.connect('/root/keel-app/api/storage/keel.db') # reuse the partner session TOK=db.execute("select token from sessions where kind='user' order by created_at desc limit 1").fetchone()[0] def call(m, path, body=None, tok=None, raw=False): req=urllib.request.Request(B+path, method=m, data=json.dumps(body).encode() if body is not None else None, headers={'Content-Type':'application/json','Authorization':'Bearer '+(tok or TOK)}) try: r=urllib.request.urlopen(req); d=r.read() except urllib.error.HTTPError as e: d=e.read(); print(' !!', m, path, e.code, d[:300].decode()); return None return d.decode() if raw else json.loads(d) def upload(name, data, cid, kind='file', tok=None): bnd='----kb'+uuid.uuid4().hex body=b'' for k,v in (('company_id',cid),('kind',kind)): body+=f'--{bnd}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode() body+=f'--{bnd}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\nContent-Type: application/octet-stream\r\n\r\n'.encode()+data+f'\r\n--{bnd}--\r\n'.encode() req=urllib.request.Request(B+'/upload', method='POST', data=body, headers={'Content-Type':f'multipart/form-data; boundary={bnd}','Authorization':'Bearer '+(tok or TOK)}) try: return json.loads(urllib.request.urlopen(req).read()) except urllib.error.HTTPError as e: print(' !! upload', e.code, e.read()[:200]); return None def png(): raw=b''.join(b'\x00'+b'\xff\xff\xff'*4 for _ in range(4)) def chunk(t,d): return struct.pack('>I',len(d))+t+d+struct.pack('>I',zlib.crc32(t+d)&0xffffffff) return b'\x89PNG\r\n\x1a\n'+chunk(b'IHDR',struct.pack('>IIBBBBB',4,4,8,2,0,0,0))+chunk(b'IDAT',zlib.compress(raw))+chunk(b'IEND',b'') mid=db.execute("select id from companies where name like 'Sahyadri%'").fetchone()[0] vid=db.execute("select id from companies where name like 'Vaigai%'").fetchone()[0] members="Folio No,Name of Member,Address,No of Shares,Cert Nos,Date of entry,Remarks\n1,Vikram Joshi,Pune,250000,\"001, 003\",10/02/2015,founder\n2,Anil Patil,Pune,200000,002,10/02/2015,\n3,Sunita Deshmukh,Mumbai,50000,004,03/03/2022,via SH-4\n" directors="Name,DIN,Designation,Appointed,Ceased\nVikas Joshi,07718842,director,10/02/2015,\nAnil Patil,07718843,director,10/02/2015,\nSunita Deshmukh,07718844,director,15/06/2019,\n" u1=upload('members_2025.csv', members.encode(), mid, 'migration'); u2=upload('directors.csv', directors.encode(), mid, 'migration'); u3=upload('minutes_scan.png', png(), mid, 'migration') print('uploads', u1 and u1['size'], u2 and u2['size'], u3 and u3['mime']) print('bad upload', upload('evil.php', b' skip; test verify route print('verify', call('POST',f'/companies/{mid}/trail/verify',{})['verify']['ok']) # public verifier on real company print('public verify', call('POST','/verify',{'reference':tr['reference']})['result'], call('POST','/verify',{'reference':'KEEL-ABCDEF-0000-123456'})['result'], call('POST','/verify',{'reference':'garbage'})) # admin db.execute("update users set role='keel' where email='p@test.test'"); db.commit() print('admin firms', [(f['name'],f['companies'],f['chains']) for f in call('GET','/admin/firms')['firms']]) print('admin chains', [(c['company'][:10],c['head'],c['anchor_to']) for c in call('GET','/admin/chains')['chains']]) print('anchor', call('POST','/admin/anchors',{'month':'2026-08'})) print('anchor dup', call('POST','/admin/anchors',{'month':'2026-08'})) print('public verify anchored', call('POST','/verify',{'reference':tr['reference']})['text'][:160]) db.execute("update users set role='partner' where email='p@test.test'"); db.commit() print('objection', call('POST','/rules/objection',{'text':'AOC-4 for Sahyadri shows 30 Oct; if the AGM day is not counted it is 31 Oct. Which is it?','rule_id':'AOC-4'})['text'][:80]) # invite + role checks print('invite bad', call('POST','/firm/invite',{'email':'nope','role':'trainee'})) print('invite', call('POST','/firm/invite',{'email':'trainee@test.test','role':'trainee','name':'T One'})) inv=db.execute("select token from invitations order by created_at desc limit 1").fetchone()[0] print('invite get', call('GET',f'/invite/{inv}')) acc=call('POST',f'/invite/{inv}',{'password':'another-long-password-1','accept_terms':True,'name':'T One'}); print('accept', acc and acc.keys()) import hmac, hashlib, time, base64 def totp(secret): key=base64.b32decode(secret+'='*((8-len(secret)%8)%8)); c=int(time.time())//30 h=hmac.new(key, struct.pack('>Q',c), hashlib.sha1).digest(); o=h[-1]&15 return '%06d' % ((struct.unpack('>I',h[o:o+4])[0]&0x7fffffff)%1000000) l=call('POST','/auth/login',{'email':'trainee@test.test','password':'another-long-password-1','code':totp(acc['totp_secret'])}); print('trainee login', l and l.keys()) if l and not l.get('token'): l=call('POST','/auth/login',{'email':'trainee@test.test','password':'another-long-password-1','code':totp(acc['totp_secret'])}) T=l['token'] print('trainee attest (should 403)', call('POST',f'/companies/{vid}/dayone/attest',{'method':'scan','upload_id':'x'},tok=T)) print('trainee invite (403)', call('POST','/firm/invite',{'email':'z@test.test'},tok=T)) print('wrong password x11:'); for i in range(11): r=call('POST','/auth/login',{'email':'trainee@test.test','password':'wrong','code':'000000'}) tid=db.execute("select id from users where email='trainee@test.test'").fetchone()[0] print('remove user', call('DELETE',f"/firm/users/{tid}")) print('trainee after removal', call('GET','/auth/me',tok=T)) print(open('/root/keel-app/php.log').read().count('Warning'), 'warnings in log')