| |
| """Bounded OOB ReadData and string codec instruction compositions.""" |
| import copy |
| import gzip |
| import itertools |
| import json |
| from pathlib import Path |
| import re |
| from analyze_area_queries import site |
| from analyze_bsp_loading import read_address |
| from analyze_bsp_references import global_address, trace_fields |
| from analyze_collision_composition import get_word, put_word |
| from analyze_engine_control import control_flow |
| from analyze_message_byte_codec import MSG, DATA |
| from analyze_message_widths import reader, run, writer |
| from analyze_message_wrappers import message_memory, SOURCE, WRAPPER_STACK |
| from analyze_native import elf |
| from analyze_qvm import write_json |
| from analyze_vm_extensions import native_functions, verified |
|
|
| DEST=0xe0000 |
|
|
|
|
| def string_graph(g): |
| g=copy.deepcopy(g);omitted=[] |
| for x in g['instructions']: |
| |
| |
| if x['opcode'] in ['pushq','popq'] or (x['opcode']=='movq' and '(%rsp)' in x['operands']): |
| omitted.append(x['address']);x['opcode']='nop';x['operands']='' |
| return g,omitted |
|
|
|
|
| def read_string(graphs,name,source,size): |
| original=graphs[name];g,omitted=string_graph(original);rows=g['instructions'];calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'};end=next(x['address'] for x in rows if x['opcode']=='retq') |
| zero=next(x for x in original['instructions'] if x['opcode']=='movb' and x['operands'].startswith('$0x0,'));base=int(re.search(r', (0x[0-9a-f]+)',zero['operands'])[1],16);capacity=8192 if name=='MSG_ReadBigString' else 1024 |
| m=message_memory(16384) |
| for i,v in enumerate(source):m[DATA+i]=v |
| m[DATA+len(source)]=0 |
| for i in range(capacity):m[base+i]=0xa5 |
| m[base-1]=m[base+capacity]=0x7b;put_word(m,MSG+28,size) |
| regs={'%rdi':MSG,'%rbx':MSG,'%rsp':WRAPPER_STACK};pc=rows[0]['address'];trace=[];count=0 |
| while True: |
| t=run(g,pc,list(calls)+[end],m,regs);m=t['memory'];regs=t['registers'];trace+=t['trace'] |
| if t['stop']==end:break |
| i=calls[t['stop']] |
| if '<MSG_ReadBits>' not in rows[i]['operands'] or [regs['rdi'],regs['rsi']]!=[MSG,8]:raise ValueError('unexpected string read request') |
| m,value,error,child=reader(graphs,m,8);trace+=child |
| if error:raise ValueError('unexpected OOB read error') |
| regs['rax']=value;count+=1;pc=rows[i+1]['address'] |
| if count>capacity:raise ValueError('string fixture request bound') |
| result=bytes(m[base+i] for i in range(capacity));length=result.index(0) |
| return {'function':name,'source_hex':bytes(source).hex(),'size':size,'capacity':capacity,'read_requests':count,'read_count':get_word(m,MSG+32),'bit':get_word(m,MSG+36),'result_hex':result[:length+1].hex(),'tail_unchanged':all(v==0xa5 for v in result[length+1:]),'returned_pointer':regs['rax'],'buffer_address':base,'canaries':[m[base-1],m[base+capacity]],'omitted_frame_instructions':omitted,**trace_fields({'trace':trace},False)} |
|
|
|
|
| def write_string(graphs,name,source,capacity,constant,strlen_override=None): |
| g=graphs[name];rows=g['instructions'];start=site(g,'testq','%rsi, %rsi');end=next(x['address'] for x in rows if x['opcode']=='movq' and re.match(r'0x(?:400|2000)\(%rsp\), %rbx',x['operands']));calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'} |
| m=message_memory(capacity);put_word(m,constant[0],constant[1],1) |
| if source is not None: |
| for i,v in enumerate(source+[0]):m[SOURCE+i]=v |
| regs={'%rsi':SOURCE if source is not None else 0,'%rbx':SOURCE if source is not None else 0,'%r12':MSG,'%rsp':WRAPPER_STACK};pc=start;trace=[];values=[];copy_requests=[];prints=0 |
| for x in graphs['MSG_WriteBits']['instructions']: |
| if x['opcode']=='addl' and '(%rip)' in x['operands']:put_word(m,global_address(x),0) |
| while True: |
| t=run(g,pc,list(calls)+[end],m,regs);m=t['memory'];regs=t['registers'];trace+=t['trace'] |
| if t['stop']==end:break |
| i=calls[t['stop']];target=rows[i]['operands'] |
| if '<strlen@plt>' in target: |
| if regs['rdi']!=SOURCE:raise ValueError('strlen pointer') |
| regs['rax']=len(source) if strlen_override is None else strlen_override |
| elif '<Q_strncpyz>' in target: |
| args=[regs[k] for k in ['rdi','rsi','rdx']];copy_requests.append(args) |
| if args[:2]!=[WRAPPER_STACK,SOURCE] or len(source)>=args[2]:raise ValueError('copy fixture') |
| for j,v in enumerate(source+[0]):m[WRAPPER_STACK+j]=v |
| elif '<Com_Printf>' in target:prints+=1 |
| elif '<MSG_WriteBits>' in target: |
| if [regs['rdi'],regs['rdx']]!=[MSG,8]:raise ValueError('string writer request') |
| values.append(regs['rsi']);m,requests,error,child=writer(graphs,m,8,regs['rsi']);trace+=child |
| if requests or error:raise ValueError('unexpected OOB write') |
| else:raise ValueError('unexpected string helper '+target) |
| pc=rows[i+1]['address'] |
| if len(values)>256:raise ValueError('string writer fixture bound') |
| size=get_word(m,MSG+28) |
| return {'function':name,'source_hex':None if source is None else bytes(source).hex(),'capacity':capacity,'supplied_strlen_word':None if source is None else len(source) if strlen_override is None else strlen_override,'requested_hex':bytes(values).hex(),'copy_requests':copy_requests,'printf_requests':prints,'payload_hex':bytes(m[DATA+i] for i in range(size)).hex(),'size':size,'bit':get_word(m,MSG+36),'overflow_flag':get_word(m,MSG+4),'source_unchanged':source is None or all(m[SOURCE+i]==v for i,v in enumerate(source+[0])),'canaries':[m[DATA-1],m[DATA+64]],**trace_fields({'trace':trace},False)} |
|
|
|
|
| def main(): |
| out={'native':{},'limitations':'Synthetic Linux ELF64 OOB instruction compositions; original processes and native callees never execute. ReadData and strings compose ReadBits/WriteBits with explicit caller-register retention and virtual frames. String strlen return and Q_strncpyz copy/NUL effects are supplied, not native library execution. Frame saves/restores/PUSH/POP excluded/stopped. Padded backing storage is distinct from cursize. No compressed strings/data, actual packets, alias/lifecycle/capacity-wrap/ABI/runtime/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());names=['MSG_ReadData','MSG_ReadString','MSG_ReadBigString','MSG_ReadStringLine','MSG_WriteString','MSG_WriteBigString','MSG_ReadBits','MSG_WriteBits'];graphs={n:control_flow(fs[n]) for n in names} |
| data_probes=[];g=graphs['MSG_ReadData'];rows=g['instructions'];start=site(g,'testl','%edx, %edx');end=site(g,'addq','$0x8, %rsp');calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'} |
| for length in [-1,0,1,2,3,4,5,6,7,8,9,16,17,31,32]: |
| for size in sorted({0,1,max(0,length-1),max(0,length)}): |
| source=[(129+37*i)&255 for i in range(40)];m=message_memory(64) |
| for i,v in enumerate(source):m[DATA+i]=v |
| for i in range(40):m[DEST+i]=0xa5 |
| m[DEST-1]=m[DEST+40]=0x7b;put_word(m,MSG+28,size) |
| regs={'%r15':length&0xffffffff,'%r14':DEST,'%rbp':MSG,'%rdx':length&0xffffffff,'%rdi':MSG,'%rsp':WRAPPER_STACK};pc=start;trace=[];count=0 |
| while True: |
| t=run(g,pc,list(calls)+[end],m,regs);m=t['memory'];regs=t['registers'];trace+=t['trace'] |
| if t['stop']==end:break |
| i=calls[t['stop']] |
| if '<MSG_ReadBits>' not in rows[i]['operands'] or [regs['rdi'],regs['rsi']]!=[MSG,8]:raise ValueError('data read request') |
| m,value,error,child=reader(graphs,m,8);trace+=child |
| if error:raise ValueError('data OOB read') |
| regs['rax']=value;count+=1;pc=rows[i+1]['address'] |
| if count>32:raise ValueError('data read fixture bound') |
| data_probes.append({'length':length,'size':size,'read_requests':count,'read_count':get_word(m,MSG+32),'bit':get_word(m,MSG+36),'source_hex':bytes(source).hex(),'destination_hex':bytes(m[DEST+i] for i in range(40)).hex(),'canaries':[m[DEST-1],m[DEST+40]],**trace_fields({'trace':trace},False)}) |
| reads=[] |
| sequences=[[0],[65,66,67,0],[37,128,255,65,10,0],[13,10,65,0],list(range(1,256))+[0]] |
| for name,source in itertools.product(['MSG_ReadString','MSG_ReadBigString','MSG_ReadStringLine'],sequences): |
| for size in sorted({0,len(source)-1,len(source)}):reads.append(read_string(graphs,name,source,size)) |
| for name in ['MSG_ReadString','MSG_ReadBigString','MSG_ReadStringLine']: |
| capacity=8192 if name=='MSG_ReadBigString' else 1024;source=[65]*capacity+[0];reads.append(read_string(graphs,name,source,len(source))) |
| writes=[];guards=[];constants={} |
| patterns=[None,[],[37,128,255,65,10]]+[[128 if i%2 else 65 for i in range(length)] for length in [1,2,3,4,5,6,7,8,9,17,31]] |
| for name in ['MSG_WriteString','MSG_WriteBigString']: |
| g=graphs[name];rows=g['instructions'];x=next(x for x in rows if x['opcode']=='movzbl' and '(%rip)' in x['operands']);address=global_address(x);value,offset=read_address(data,meta['sections'],address,1);constant=(address,value[0]);constants[name]={'address':address,'file_offset':offset,'byte':value[0]} |
| for source,capacity in itertools.product(patterns,[4,64]):writes.append(write_string(graphs,name,source,capacity,constant)) |
| ci=next(i for i,x in enumerate(rows) if '<strlen@plt>' in x['operands']);stops=[x['address'] for x in rows if x['opcode']=='callq' and ('<Q_strncpyz>' in x['operands'] or '<Com_Printf>' in x['operands'])];limit=8192 if name=='MSG_WriteBigString' else 1024 |
| for length in [limit,limit+1]:writes.append(write_string(graphs,name,[],64,constant,strlen_override=length)) |
| for length in [0,limit-1,limit,limit+1,0xffffffff,0x80000000]: |
| t=run(g,rows[ci+1]['address'],stops,{}, {'%rax':length,'%rbx':SOURCE,'%rsp':WRAPPER_STACK});target=next(x['operands'] for x in rows if x['address']==t['stop']);guards.append({'function':name,'supplied_strlen_word':length,'outcome':'copy_request' if '<Q_strncpyz>' in target else 'printf_request','copy_arguments':[t['registers'][k] for k in ['rdi','rsi','rdx']] if '<Q_strncpyz>' in target else None,**trace_fields(t,False)}) |
| out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,'data_reads':data_probes,'string_reads':reads,'string_writes':writes,'writer_length_guards':guards,'writer_empty_constants':constants} |
| write_json(Path('docs/research/message-strings.json'),out) |
| print('Recovered OOB data reads and string read/write boundary compositions') |
|
|
|
|
| if __name__=='__main__':main() |
|
|
