#!/usr/bin/env python3 """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) # Padding bytes model local memory access, not a valid message length. 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 '' in name: error={'arguments':[regs[k] for k in ['rdi','rsi','rdx']]};break if '' 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 '' 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) # Huffman tree decoding remains stopped; symbol and variable code length supplied. 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()