#!/usr/bin/env python3 """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']: # Explicit frame-save/restore omission, including the line reader's # tail restores which precede its final terminator store. 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 '' 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 '' in target: if regs['rdi']!=SOURCE:raise ValueError('strlen pointer') regs['rax']=len(source) if strlen_override is None else strlen_override # Explicit supplied library return. elif '' 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 # Supplied copy/NUL effects. elif '' in target:prints+=1 # Logging callee remains unexecuted. elif '' 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 '' 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 '' in x['operands']);stops=[x['address'] for x in rows if x['opcode']=='callq' and ('' in x['operands'] or '' 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 '' in target else 'printf_request','copy_arguments':[t['registers'][k] for k in ['rdi','rsi','rdx']] if '' 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()