| |
| """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() |
|
|
