File size: 11,303 Bytes
e517ae4 | 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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | #!/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 '<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 # Explicit supplied library return.
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 # Supplied copy/NUL effects.
elif '<Com_Printf>' in target:prints+=1 # Logging callee remains unexecuted.
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()
|