File size: 5,847 Bytes
6dafc4b | 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 | #!/usr/bin/env python3
"""Bounded Linux client mask guard and reader slices; bit returns are supplied."""
import copy
import gzip
import itertools
import json
from pathlib import Path
from analyze_area_queries import site
from analyze_bsp_references import trace_fields
from analyze_chat_paths import native_string
from analyze_native import elf
from analyze_collision_composition import evaluate,put_word
from analyze_engine_control import control_flow
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 data_probe(g,length,size,frame=0x20000):
rows=g['instructions'];calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'}
end=site(g,'addq','$0x8, %rsp');message=0x10000;dest=frame+24
m={frame+i:0xa5 for i in range(-1,541)}
# The CL_ParseSnapshot memset is represented by explicitly supplied effects.
for i in range(540):m[frame+i]=0
put_word(m,message+28,size);put_word(m,message+32,0)
regs={'%r15':length&0xffffffff,'%r14':dest,'%rbp':message,'%rdx':length&0xffffffff,'%rdi':message}
start=site(g,'testl','%edx, %edx');trace=[];requests=[];returns=[]
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']]
if '<MSG_ReadBits>' not in rows[i]['operands']:raise ValueError('unexpected helper')
requests.append([regs['rdi'],regs['rsi']]);ordinal=len(requests)
value=[0,127,128,255,0x1234,0xffffffff][(ordinal-1)%6]
returns.append(value);put_word(m,message+32,ordinal);regs['rax']=value
start=rows[i+1]['address']
if ordinal>32:raise ValueError('fixture request bound')
return {'length':length,'size':size,'supplied_bits_returns':returns,'supplied_post_readcounts':list(range(1,len(returns)+1)),
'requests':requests,'mask_bytes':[m[dest+i] for i in range(32)],'canaries':[m[frame-1],m[frame+540]],**trace_fields({'trace':trace},True)}
def main():
label='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']);metadata,_=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 ['CL_ParseSnapshot','MSG_ReadByte','MSG_ReadData']}
parse=graphs['CL_ParseSnapshot'];rows=parse['instructions'];dest_call=next(x['address'] for x in rows if '<MSG_ReadData>' in x['operands']);error=next(x['address'] for x in rows if '<Com_Error>' in x['operands'])
starts=[x['address'] for x in rows if x['opcode']=='cmpl' and x['operands'].split('#')[0].strip()=='$0x20, %eax']
guards=[]
for start,length in itertools.product(starts,[0,1,7,31,32,33,127,128,255,0xffffffff,0x80000000]):
t=run(parse,start,[dest_call,error],{}, {'%rax':length,'%r12':0x20000,'%rbx':0x10000})
regs=t['registers'];guards.append({'start':start,'length_word':length,'accepted':t['stop']==dest_call,'arguments':[regs['rdi'],regs['rsi'],regs['rdx']],**trace_fields(t,True)})
reset_index=next(i for i,x in enumerate(rows) if '<memset@plt>' in x['operands'])
# Frame setup and pushes are excluded, but native request operands are evaluated.
reset=copy.deepcopy(parse)
omitted=[]
for x in reset['instructions'][:reset_index]:
if x['opcode']=='pushq':omitted.append(x['address']);x['opcode']='nop';x['operands']=''
t=run(reset,rows[0]['address'],[rows[reset_index]['address']],{}, {'%rsp':0x20220,'%rdi':0x10000})
byte=copy.deepcopy(graphs['MSG_ReadByte']);br=byte['instructions'];call_index=next(i for i,x in enumerate(br) if '<MSG_ReadBits>' in x['operands']);ret=next(x['address'] for x in br if x['opcode']=='retq')
byte_omitted=[]
for x in br:
if x['opcode']=='popq':byte_omitted.append(x['address']);x['opcode']='nop';x['operands']=''
byte_probes=[]
for size,cursor,value in itertools.product([0,1,32,0x7fffffff,0xffffffff],[0,1,2,32,33,0x80000000,0xffffffff],[0,127,128,255,0x1234,0xffffffff]):
m={};put_word(m,0x10000+28,size);put_word(m,0x10000+32,cursor)
b=run(byte,br[call_index+1]['address'],[ret],m,{'%rbx':0x10000,'%rax':value})
byte_probes.append({'size_word':size,'cursor_word':cursor,'supplied_bits_return':value,'result_word':b['registers']['rax'],**trace_fields(b,False)})
probes=[data_probe(graphs['MSG_ReadData'],length,size) 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)})]
out={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,
'reset_request':{'arguments':[t['registers'][k] for k in ['rdi','rsi','rdx']],'omitted_push_addresses':omitted,**trace_fields(t,True)},
'error_format':native_string(data,metadata['sections'],0x569928),'guard_probes':guards,'byte_probes':byte_probes,'byte_omitted_pop_addresses':byte_omitted,'data_probes':probes,
'limitations':'Linux client only. Original process and every native callee remain unexecuted. Frame PUSH/POP omitted explicitly; virtual stack supplied. Snapshot memset effects, MSG_ReadBits returns and post-call readcounts supplied, not encoded/decoded packets. No full CL_ParseSnapshot execution, real Huffman/bit cursor, ABI, delta/player/entity validity, publication or renderer conformance.'}
write_json(Path('docs/research/client-masks.json'),out)
print(f"Recovered {len(guards)} guards, {len(byte_probes)} byte-reader and {len(probes)} data-reader compositions")
if __name__=='__main__':main()
|