| |
| """Synthetic native read-mode, OOB and compressed reader compositions.""" |
| import gzip |
| import itertools |
| import json |
| from pathlib import Path |
| from analyze_area_queries import site |
| from analyze_bsp_references import global_address,trace_fields |
| from analyze_chat_paths import native_string |
| from analyze_collision_composition import evaluate,get_word,put_word |
| from analyze_engine_control import control_flow |
| from analyze_native import elf |
| from analyze_qvm import write_json |
| from analyze_vm_extensions import native_functions,verified |
|
|
|
|
| def run(g,start,stops,m,regs):return evaluate(g,start,stops,m,regs,limit=10000,capture_registers=True) |
|
|
|
|
| def reader(graphs,width,mode,pattern,cursor=0,initial_bit=0,size=0): |
| g=graphs['MSG_ReadBits'];rows=g['instructions'];message,data,stack=0x10000,0x20000,0x30000 |
| start=site(g,'testl','%esi, %esi');end=next(x['address'] for x in rows if x['opcode']=='movl' and x['operands'].split('#')[0].strip()=='%ebp, %eax') |
| calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'} |
| m={} |
| for x in rows: |
| if x['opcode']=='movq' and '(%rip)' in x['operands']:put_word(m,global_address(x),0x40000,8) |
| put_word(m,message+8,mode);put_word(m,message+16,data,8);put_word(m,message+28,size);put_word(m,message+32,cursor);put_word(m,message+36,initial_bit) |
| |
| for i in range(-8,24):m[data+i]=[0,255,0x96,0x81][pattern] |
| regs={'%r12':message,'%r15':width&0xffffffff,'%rsi':width&0xffffffff,'%rsp':stack} |
| trace=[];requests=[];symbols=[];raw_bits=[];children=[];error=None |
| while True: |
| t=run(g,start,list(calls)+[end],m,regs);trace+=t['trace'];m=t['memory'];regs=t['registers'] |
| if t['stop']==end:break |
| i=calls[t['stop']];name=rows[i]['operands'] |
| if '<Com_Error>' in name: |
| error={'arguments':[regs[k] for k in ['rdi','rsi','rdx']]};break |
| if '<Huff_getBit>' in name: |
| requests.append(['Huff_getBit',regs['rdi'],regs['rsi']]);h=graphs['Huff_getBit'];ret=next(x['address'] for x in h['instructions'] if x['opcode']=='retq') |
| child=run(h,h['instructions'][0]['address'],[ret],m,{'%rdi':regs['rdi'],'%rsi':regs['rsi']});m=child['memory'];regs['rax']=child['registers']['rax'];raw_bits.append(regs['rax']);children+=child['trace'] |
| elif '<Huff_offsetReceive>' in name: |
| requests.append(['Huff_offsetReceive',regs['rdi'],regs['rsi'],regs['rdx'],regs['rcx']]) |
| symbol=([0,255,0x96,0x81][pattern]+37*len(symbols))&255;symbols.append(symbol) |
| |
| put_word(m,regs['rsi'],symbol);put_word(m,regs['rcx'],get_word(m,regs['rcx'])+5+len(symbols)%4) |
| else:raise ValueError('unexpected helper '+name) |
| if len(requests)>36:raise ValueError('fixture request bound') |
| start=rows[i+1]['address'] |
| return {'width':width,'mode_word':mode,'pattern':pattern,'initial_readcount':cursor,'initial_bit':initial_bit,'cursize':size, |
| 'result_word':None if error else regs['rbp']&0xffffffff,'error':error,'final_readcount':get_word(m,message+32),'final_bit':get_word(m,message+36), |
| 'remaining_width_word':regs['r15']&0xffffffff,'raw_bits':raw_bits,'supplied_symbols':symbols,'requests':requests, |
| 'final_raw_bit_global':get_word(m,global_address(next(x for x in graphs['Huff_getBit']['instructions'] if '(%rip)' in x['operands']))) if raw_bits else None, |
| 'child_trace':trace_fields({'trace':children},False),**trace_fields({'trace':trace},False)} |
|
|
|
|
| def main(): |
| out={'native':{},'limitations':'Two Linux ELF64 engines, bounded synthetic read slices only. Original processes/callees never execute. Prologue saves/restores excluded and virtual stack supplied. Huff_getBit instructions composed with explicit caller-register restoration; Huff_offsetReceive symbol/cursor effects supplied. Padded backing memory is distinct from cursize. No real Huffman tree/writer/packet/error execution, ABI, legal malformed-state reachability or all-platform conformance.'} |
| for label in ['linux--Quake3-UrT-Ded.x86_64','linux--Quake3-UrT.x86_64']: |
| r=json.loads(Path(f'docs/research/native/{label}.json').read_text());data=verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(),r['source']['sha256']);meta,_=elf(data) |
| a=r['artifacts']['disassembly.txt'];fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode());graphs={n:control_flow(fs[n]) for n in ['MSG_BeginReading','MSG_BeginReadingOOB','MSG_ReadBits','Huff_getBit']} |
| resets=[] |
| for name,initial in itertools.product(['MSG_BeginReading','MSG_BeginReadingOOB'],[0,0xa5]): |
| g=graphs[name];m={0x10000+i:initial for i in range(-1,41)};end=next(x['address'] for x in g['instructions'] if x['opcode']=='retq');t=run(g,g['instructions'][0]['address'],[end],m,{'%rdi':0x10000}) |
| resets.append({'function':name,'initial_byte':initial,'record_bytes':[t['memory'][0x10000+i] for i in range(40)],'canaries':[t['memory'][0xffff],t['memory'][0x10028]],**trace_fields(t,True)}) |
| oob=[reader(graphs,w,mode,p,cursor,19,size) for w,mode,p,cursor,size in itertools.product([8,16,32,-8,-16,-32,0,1,7,9,24,33],[1],[0,1,2,3],[-1,0,3],[0,8])] |
| oob += [reader(graphs,w,2,p,0,19,0) for w,p in itertools.product([8,16,32,0,7,-8],[1,2])] |
| compressed=[reader(graphs,w,0,p,0,bit,0) for w,p,bit in itertools.product(list(range(0,33))+[-1,-7,-8,-9,-15,-16,-17,-24,-31,-32],[0,1,2,3],[0,7])] |
| g=graphs['MSG_ReadBits'];err=next(x for x in g['instructions'] if x['opcode']=='movl' and x['operands'].split('#')[0].strip().endswith(', %esi') and '$0x' in x['operands']);addr=int(err['operands'].split(',')[0][1:],0) |
| bit_g=graphs['Huff_getBit'];bloc=global_address(next(x for x in bit_g['instructions'] if '(%rip)' in x['operands'])) |
| out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,'error_format':native_string(data,meta['sections'],addr),'bit_global_address':bloc,'reset_probes':resets,'oob_probes':oob,'compressed_probes':compressed} |
| write_json(Path('docs/research/message-bits.json'),out) |
| print('Recovered eight resets, 600 OOB and 688 compressed reader compositions') |
|
|
|
|
| if __name__=='__main__':main() |
|
|
