File size: 7,406 Bytes
0ef6dab | 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 | #!/usr/bin/env python3
"""Recover Linux ELF64 spatial/trig dispatch and bounded VM pointer traces."""
import gzip
import json
from pathlib import Path
import re
import struct
from analyze_chat_paths import native_string
from analyze_deferred_auth import signed32
from analyze_engine_control import control_flow
from analyze_frame_timing import read_address
from analyze_native import elf
from analyze_native_dispatch import recover
from analyze_qvm import write_json
from analyze_vm_extensions import native_functions,verified
MASK64=2**64-1
def case_path(graph,address,limit=128):
index=next((i for i,x in enumerate(graph['instructions']) if x['address']==address),None)
if index is None:raise ValueError('case address missing')
rows=[]
for x in graph['instructions'][index:index+limit]:
rows.append(x)
if x['opcode'] in ['jmp','jmpq','ret','retq']:return rows
if x['opcode'].startswith('j'):raise ValueError('branching case needs a full CFG')
raise ValueError('case lacks bounded terminal transfer')
def pointer_trace(graph,argument,current_vm,flag,data_base,data_mask,limit=64):
"""Evaluate only the observed VM_ArgPtr integer instructions with synthetic words."""
instructions={x['address']:x for x in graph['instructions']}
order={x['address']:i for i,x in enumerate(graph['instructions'])};rows=graph['instructions']
regs={'%rdi':argument&MASK64};m={current_vm+0x58:flag,current_vm+0x90:data_base,
current_vm+0x98:data_mask};zero=False;trace=[];pc=graph['entry']
def reg(r):
if r not in regs:raise ValueError('unknown pointer register')
return regs[r]
def read(address):
if address not in m:raise ValueError('unknown pointer word')
return m[address]
for _ in range(limit):
if pc not in instructions:raise ValueError('pointer trace escaped function')
x=instructions[pc];trace.append(pc);op=x['opcode'];args=x['operands'].split('#')[0].strip()
nextpc=rows[order[pc]+1]['address'] if order[pc]+1<len(rows) else None
if op=='testq':
a,b=args.split(', ');zero=(reg(a)®(b))==0
elif op=='xorl' and args=='%eax, %eax':regs['%rax']=0
elif op in ['je','jne']:
if zero==(op=='je'):nextpc=int(args.split()[0],16)
elif op=='cmpq' and args=='$0x0, 0x58(%rdx)':zero=read(reg('%rdx')+0x58)==0
elif op=='movq' and '(%rip), %rdx' in args:
if '<currentVM>' not in x['operands']:raise ValueError('pointer global identity changed')
regs['%rdx']=current_vm
elif op=='movq' and args=='%rdi, %rax':regs['%rax']=reg('%rdi')
elif op=='movslq' and args=='0x98(%rdx), %rax':regs['%rax']=signed32(read(reg('%rdx')+0x98))&MASK64
elif op=='andq' and args=='%rdi, %rax':regs['%rax']=reg('%rax')®('%rdi')
elif op=='addq' and args=='0x90(%rdx), %rax':regs['%rax']=(reg('%rax')+read(reg('%rdx')+0x90))&MASK64
elif op in ['ret','retq']:return {'trace':trace,'result':reg('%rax')}
else:raise ValueError('unsupported pointer instruction '+op+' '+args)
pc=nextpc
raise ValueError('pointer trace step limit')
def main():
out={'native':{},'limitations':'Static Linux ELF64 dispatch/graphs and bounded synthetic VM_ArgPtr '
'instruction traces. No original process/native callee/collision/trig service executes. Pointer '
'fixtures supply currentVM fields; allocation/mask invariants, malformed pointers, exceptional '
'floats, geometry/portal behavior and other platforms remain open.'}
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'])
a=r['artifacts']['disassembly.txt'];text=verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode()
metadata,_=elf(data);fs=native_functions(text);dispatch=recover(data,text,'SV_GameSystemCalls')
graph=control_flow(fs['SV_GameSystemCalls']);cases={}
for n in [24,26,32,103,104]:
case=next(c for c in dispatch['cases'] if n in c['numbers_sharing_target'])
path=case_path(graph,case['target_address'])
calls=[x for x in path if x['opcode'].startswith('call')]
cases[str(n)]={'target_address':case['target_address'],'instructions':path,'calls':calls}
names=['VM_ArgPtr','SV_Trace','SV_inPVS','SV_AreaEntities','SV_AreaEntities_r','SV_ClipMoveToEntities']
graphs={n:control_flow(fs[n]) for n in names}
zero_ins=next(x for x in graphs['SV_Trace']['instructions'] if re.match(r'\$0x[0-9a-f]+, %esi',x['operands']))
zero_address=int(re.search(r'\$0x([0-9a-f]+)',zero_ins['operands'])[1],16)
sections=[s for s in metadata['sections'] if s['address']<=zero_address and zero_address+12<=s['address']+s['bytes']]
if len(sections)!=1:raise ValueError('ambiguous extent section')
section=sections[0]
if section['zero_fill']:raw,offset=bytes(12),None
else:raw,offset=read_address(data,metadata['sections'],zero_address,12)
constants=[]
for x in graphs['SV_Trace']['instructions']:
if x['opcode']=='movss' and '(%rip)' in x['operands']:
address=int(re.search(r'#\s*0x([0-9a-f]+)',x['operands'])[1],16)
value,fileoff=read_address(data,metadata['sections'],address,4)
constants.append({'instruction':x['address'],'address':address,'file_offset':fileoff,
'bits':struct.unpack('<I',value)[0],'float32':struct.unpack('<f',value)[0]})
warning_ins=next(x for x in reversed(graphs['SV_AreaEntities_r']['instructions'])
if x['opcode']=='movl' and re.match(r'\$0x[0-9a-f]+, %edi',x['operands']))
warning_address=int(re.search(r'\$0x([0-9a-f]+)',warning_ins['operands'])[1],16)
warning={'address':warning_address,'text':native_string(data,metadata['sections'],warning_address)}
traces=[]
for argument,vm,flag,mask in [(0,0,0,65535),(1,0,0,65535),(256,0x20000,1,65535),
(0x1234,0x20000,0,255),(511,0x20000,0,255),(-1,0x20000,0,65535),
(0x1234,0x20000,0,-1),(2**32,0x20000,1,65535),(2**32,0x20000,0,65535)]:
t=pointer_trace(graphs['VM_ArgPtr'],argument,vm,flag,0x100000,mask)
traces.append({'argument':argument,'current_vm':vm,'field58':flag,'data_base':0x100000,
'mask_field98':mask,'trace':t})
out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],
'dispatch_table_address':dispatch['table_address'],'cases':cases,'graphs':graphs,
'zero_extent':{'address':zero_address,'file_offset':offset,'section':section['name'],
'zero_fill_initial_value':section['zero_fill'],'raw':raw.hex(),
'float32':list(struct.unpack('<3f',raw))},'trace_float_constants':constants,
'pointer_traces':traces,'area_capacity_warning':warning}
write_json(Path('docs/research/native-spatial.json'),out)
print('Recovered paired Linux trig/spatial dispatch and 18 bounded VM pointer paths')
if __name__=='__main__':main()
|