Download scripts/analyze_area_queries.py from nmarshall/urt-rs: direct link, hf CLI and curl.
- Browser
- Download file 8.51 kB
-
https://huggingface.co/nmarshall/urt-rs/resolve/main/scripts/analyze_area_queries.py
- Command line
-
hf download hf://nmarshall/urt-rs/scripts/analyze_area_queries.py
-
curl -L -o analyze_area_queries.py https://huggingface.co/nmarshall/urt-rs/resolve/main/scripts/analyze_area_queries.py
8.51 kB
| #!/usr/bin/env python3 | |
| """Bounded entity addressing, area overlap/insertion and split traversal evidence.""" | |
| import gzip | |
| import itertools | |
| import json | |
| from pathlib import Path | |
| import re | |
| from analyze_chat_crosshair import bits | |
| from analyze_collision_composition import evaluate, get_word, put_word | |
| from analyze_engine_control import control_flow | |
| from analyze_qvm import write_json | |
| from analyze_vm_extensions import native_functions, verified | |
| def site(g,op,args): | |
| matches=[x['address'] for x in g['instructions'] if x['opcode']==op and x['operands'].split('#')[0].strip()==args] | |
| if len(matches)!=1:raise ValueError('ambiguous area site '+args) | |
| return matches[0] | |
| def run(g,start,stops,m,regs):return evaluate(g,start,stops,m,regs,capture_registers=True) | |
| def main(): | |
| out={'native':{},'limitations':'Bounded synthetic instructions with supplied pointers, registered fields, ' | |
| 'bounds/counts and recursive-return state. No original process/native callee executes. Finite floats ' | |
| 'only; linked-list/tree construction, actual query composition, pointer validity, ABI lifetimes, ' | |
| 'geometry and original-runtime/other-platform behavior 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()) | |
| verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(),r['source']['sha256']) | |
| 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 ['SV_GentityNum','SV_GEntityForSvEntity','SV_LocateGameData','SV_AreaEntities','SV_AreaEntities_r']} | |
| def global_address(n,op): | |
| x=next(x for x in graphs[n]['instructions'] if x['opcode']==op and '(%rip)' in x['operands']) | |
| return int(re.search(r'#\s*0x([0-9a-f]+)',x['operands'])[1],16) | |
| stride_address=global_address('SV_GentityNum','imull');base_address=global_address('SV_GentityNum','addq') | |
| g=graphs['SV_GEntityForSvEntity'];svbase=int(next(x for x in g['instructions'] if x['opcode']=='subq')['operands'].split(',')[0][1:],0) | |
| entity=[] | |
| for index,stride in itertools.product([0,1,7,1023,1024,0xffffffff,0x7fffffff,0x80000000],[880,312,0,0xffffffff]): | |
| m={};put_word(m,stride_address,stride);put_word(m,base_address,0x100000000,8) | |
| g=graphs['SV_GentityNum'];stop=next(x['address'] for x in g['instructions'] if x['opcode']=='retq') | |
| t=run(g,g['entry'],[stop],m,{'%rdi':index}) | |
| entity.append({'index_bits':index,'stride_bits':stride,'base':0x100000000,'result':t['registers']['rax'],'trace':t['trace']}) | |
| reverse=[] | |
| g=graphs['SV_GEntityForSvEntity'];stop=next(x['address'] for x in g['instructions'] if x['opcode']=='retq') | |
| for index,stride in itertools.product([-1,0,1,7,1023,1024],[880,312]): | |
| m={};put_word(m,stride_address,stride);put_word(m,base_address,0x100000000,8) | |
| t=run(g,g['entry'],[stop],m,{'%rdi':svbase+312*index}) | |
| reverse.append({'sv_record_index':index,'sv_pointer':svbase+312*index,'stride':stride,'base':0x100000000,'result':t['registers']['rax'],'trace':t['trace']}) | |
| g=graphs['SV_LocateGameData'];stop=next(x['address'] for x in g['instructions'] if x['opcode']=='retq') | |
| registration=run(g,g['entry'],[stop],{}, {'%rdi':0x100000000,'%rsi':1024,'%rdx':880,'%rcx':0x200000000,'%r8':256}) | |
| g=graphs['SV_AreaEntities_r'];rows=g['instructions'] | |
| overlap_start=site(g,'movq','0x8(%rbp), %rdx');accept=site(g,'movl','0x18(%rbp), %eax');reject=site(g,'testq','%r12, %r12') | |
| count_cmp=site(g,'cmpl','0x1c(%rbp), %eax');i=next(i for i,x in enumerate(rows) if x['address']==count_cmp) | |
| if rows[i+1]['opcode']!='je':raise ValueError('capacity branch changed') | |
| warning=int(rows[i+1]['operands'].split()[0],16) | |
| split_sites=[i for i,x in enumerate(rows[:-1]) if x['opcode']=='movl' and x['operands']=='(%r13), %ecx' and rows[i+1]['operands']=='$-0x1, %ecx'] | |
| if len(split_sites)!=1:raise ValueError('ambiguous split site') | |
| i=split_sites[0];split_start=rows[i]['address'] | |
| if rows[i+2]['opcode']!='je':raise ValueError('leaf branch changed') | |
| leaf_exit=int(rows[i+2]['operands'].split()[0],16) | |
| tail=site(g,'movq','0x18(%r13), %rbx') | |
| recursive=[i for i,x in enumerate(rows) if '<SV_AreaEntities_r>' in x['operands'] and x['opcode']=='callq'] | |
| if len(recursive)!=1:raise ValueError('recursive site changed') | |
| call=rows[recursive[0]]['address'];after_call=rows[recursive[0]+1]['address'] | |
| query,obj,lo,hi,output,node=0x20000,0x30000,0x40000,0x40020,0x50000,0x60000 | |
| def query_memory(minimum,maximum): | |
| m={};put_word(m,query,lo,8);put_word(m,query+8,hi,8);put_word(m,query+16,output,8) | |
| for axis in range(3):put_word(m,lo+4*axis,bits(minimum[axis]));put_word(m,hi+4*axis,bits(maximum[axis])) | |
| return m | |
| overlap=[] | |
| choices=[(-2,-1),(-1,1),(1,2),(1.01,2),(-2,-1.01),(-.5,.5)] | |
| for bounds in itertools.product(choices,repeat=3): | |
| m=query_memory([-1]*3,[1]*3) | |
| for axis,(minimum,maximum) in enumerate(bounds):put_word(m,obj+0x1d0+4*axis,bits(minimum));put_word(m,obj+0x1dc+4*axis,bits(maximum)) | |
| t=run(g,overlap_start,[accept,reject],m,{'%rbp':query,'%rax':obj}) | |
| overlap.append({'entity_min':[x[0] for x in bounds],'entity_max':[x[1] for x in bounds], 'accepted':t['stop']==accept,'trace':t['trace']}) | |
| insertion=[] | |
| magic=next(x for x in rows if x['opcode']=='movabsq')['operands'].split(',')[0][1:] | |
| for count,capacity in [(0,0),(0,1),(1,1),(1,2),(1023,1024),(1024,1024),(1025,1024),(-1,1024)]: | |
| for index in [0,1,1023]: | |
| m={output-1:0xab,output+4096:0xcd};put_word(m,query+16,output,8);put_word(m,query+24,count);put_word(m,query+28,capacity) | |
| t=run(g,accept,[reject,warning],m,{'%rbp':query,'%rbx':svbase+312*index,'%r14':int(magic,0)}) | |
| insertion.append({'count_before':count,'capacity':capacity,'sv_record_index':index,'warning_path':t['stop']==warning, | |
| 'count_after':get_word(t['memory'],query+24),'output_writes':[{'address':w['address'],'bytes':w['bytes'],'value':get_word(t['memory'],w['address'],w['bytes'])} for w in t['writes'] if w['address']!=query+24], | |
| 'trace':t['trace']}) | |
| traversal=[] | |
| for axis in [-1,0,1,2]: | |
| for minimum,maximum in [(-2,-1),(-1,0),(0,0),(0,1),(-1,1),(1,2)]: | |
| m=query_memory([minimum]*3,[maximum]*3);put_word(m,node,axis);put_word(m,node+4,bits(0));put_word(m,node+8,node+0x100,8);put_word(m,node+16,node+0x200,8) | |
| regs={'%r13':node,'%rbp':query};t=run(g,split_start,[call,tail,leaf_exit],m,regs);trace=t['trace'];children=[] | |
| if t['stop']==call: | |
| children.append(0) | |
| # Isolated return boundary: no recursive callee runs; preserved node/query are supplied. | |
| t=run(g,after_call,[tail,leaf_exit],m,{'%r13':node,'%rbp':query});trace+=t['trace'] | |
| if t['stop']==tail:children.append(1) | |
| traversal.append({'axis':axis,'query_min':minimum,'query_max':maximum,'split':0,'children':children,'trace':trace}) | |
| out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs, | |
| 'globals':{'entity_base':base_address,'entity_stride':stride_address,'sv_record_base':svbase}, | |
| 'entity_address_traces':entity,'sv_to_game_address_traces':reverse, | |
| 'registration':{'registers':registration['registers'],'trace':registration['trace'],'writes':[{'address':w['address'],'bytes':w['bytes'],'value':get_word(registration['memory'],w['address'],w['bytes'])} for w in registration['writes']]}, | |
| 'sites':{'overlap':overlap_start,'accept':accept,'reject':reject,'capacity_warning':warning,'split':split_start,'child0_call':call,'after_child0_call':after_call,'child1_tail':tail,'leaf_exit':leaf_exit}, | |
| 'overlap_traces':overlap,'insertion_traces':insertion,'split_traversal_compositions':traversal} | |
| write_json(Path('docs/research/area-queries.json'),out) | |
| print('Recovered 64 entity/24 reverse-address, 432 overlap, 48 insertion and 48 split cases') | |
| if __name__=='__main__':main() | |