File size: 6,437 Bytes
6787ca3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | #!/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 '<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)
# 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()
|