urt-rs / scripts /analyze_message_numeric.py
nmarshall's picture
Recover float transport and finite angle message codecs
7e134bc
Raw History Blame
8.11 kB
#!/usr/bin/env python3
"""Raw float transport and bounded finite angle instruction compositions."""
import copy
import gzip
import itertools
import json
from pathlib import Path
import re
from analyze_bsp_loading import read_address
from analyze_bsp_references import global_address, trace_fields
from analyze_chat_crosshair import bits
from analyze_collision_composition import get_word, put_word
from analyze_engine_control import control_flow
from analyze_message_byte_codec import MSG
from analyze_message_widths import reader, run, writer
from analyze_message_wrappers import message_memory, WRAPPER_STACK
from analyze_native import elf
from analyze_qvm import write_json
from analyze_vm_extensions import native_functions, verified
WORDS=[0,0x80000000,0x3f800000,0xbf800000,0x3fc00000,0x7f7fffff,1,0x7f800000,0xff800000,0x7fc01234,0x7f801234,0xffffffff]
ANGLES=[0.0,-0.0,1.0,-1.0,45.125,-45.125,90.0,-90.0,180.0,270.0,359.9,360.0,-360.0,720.0,65535.0]
def frame_graph(g):
g=copy.deepcopy(g);omitted=[]
for x in g['instructions']:
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_float(g,m,value):
g,omitted=frame_graph(g);rows=g['instructions'];ci=next(i for i,x in enumerate(rows) if '<MSG_ReadBits>' in x['operands']);end=next(x['address'] for x in rows if x['opcode']=='retq')
# Start after the helper: supply its return and retained message/frame.
t=run(g,rows[ci+1]['address'],[end],m,{'%rax':value,'%rbx':MSG,'%rsp':WRAPPER_STACK})
return t['registers']['xmm0'],t['trace'],omitted
def read_angle(g,m,value):
g,omitted=frame_graph(g);rows=g['instructions'];ci=next(i for i,x in enumerate(rows) if '<MSG_ReadBits>' in x['operands']);ends=[x['address'] for x in rows if x['opcode']=='retq']
t=run(g,rows[ci+1]['address'],ends,m,{'%rax':value,'%rbx':MSG})
return t['registers']['xmm0'],t['trace'],omitted
def main():
out={'native':{},'limitations':'Synthetic Linux ELF64 instruction compositions only. Original processes and native callees never execute. Float transport evaluates raw moves, including nonfinite payload words, not floating arithmetic. Angle arithmetic is restricted to supplied finite in-range inputs, signed int32 conversion, scalar binary64/32 operations under modeled ordinary rounding; exception/MXCSR/FTZ/DAZ/NaN/overflow behavior unmodeled. OOB child writer/reader instructions composed with explicit caller-register retention; frame saves/restores/PUSH/POP omitted/stopped. Supplied tail return/count probes are distinct from actual OOB compositions. Compressed/full-packet/runtime/all-platform conformance remains 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']);meta,_=elf(data);a=r['artifacts']['disassembly.txt'];fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode());names=['MSG_WriteFloat','MSG_ReadFloat','MSG_WriteAngle','MSG_WriteAngle16','MSG_ReadAngle16','MSG_WriteBits','MSG_ReadBits'];graphs={n:control_flow(fs[n]) for n in names};constants=[];constant_memory={}
for name in ['MSG_WriteAngle','MSG_WriteAngle16','MSG_ReadAngle16']:
for x in graphs[name]['instructions']:
if '(%rip)' not in x['operands']:continue
size=8 if x['opcode']=='mulsd' else 4;address=global_address(x);raw,offset=read_address(data,meta['sections'],address,size)
for i,v in enumerate(raw):constant_memory[address+i]=v
constants.append({'function':name,'opcode':x['opcode'],'address':address,'file_offset':offset,'bytes_hex':raw.hex()})
floats=[];tails=[]
wg=graphs['MSG_WriteFloat'];jump=next(x['address'] for x in wg['instructions'] if x['opcode']=='jmp')
for word,capacity in itertools.product(WORDS,[3,64]):
t=run(wg,wg['instructions'][0]['address'],[jump],{}, {'%rdi':MSG,'%rsp':WRAPPER_STACK,'%xmm0':word});args=[t['registers'][k] for k in ['rdi','rsi','rdx']];m=message_memory(capacity)
for x in graphs['MSG_WriteBits']['instructions']:
if x['opcode']=='addl' and '(%rip)' in x['operands']:put_word(m,global_address(x),0)
m,requests,error,wt=writer(graphs,m,args[2],args[1]);size=get_word(m,MSG+28);bit=get_word(m,MSG+36);payload=bytes(m[0xa0000+i] for i in range(size)).hex();flag=get_word(m,MSG+4);put_word(m,MSG+32,0);put_word(m,MSG+36,7)
m,raw,read_error,rt=reader(graphs,m,32);result,tail,omitted=read_float(graphs['MSG_ReadFloat'],m,raw)
floats.append({'input_word':word,'capacity':capacity,'write_arguments':args,'payload_hex':payload,'write_size':size,'write_bit':bit,'read_count':get_word(m,MSG+32),'read_bit':get_word(m,MSG+36),'raw_read_word':raw,'result_word':result,'overflow_flag':flag,'omitted_frame_instructions':omitted,**trace_fields({'trace':t['trace']+wt+rt+tail},False)})
if requests or error or read_error:raise ValueError('unexpected float OOB helper')
for word,size,count in itertools.product(WORDS,[0,4,0xffffffff],[0,4,5,0x80000000,0xffffffff]):
m={};put_word(m,MSG+28,size);put_word(m,MSG+32,count);result,trace,omitted=read_float(graphs['MSG_ReadFloat'],m,word);tails.append({'supplied_word':word,'size_word':size,'post_count_word':count,'result_word':result,**trace_fields({'trace':trace},False)})
writes=[]
for name,angle in itertools.product(['MSG_WriteAngle','MSG_WriteAngle16'],ANGLES):
g=graphs[name];jump=next(x['address'] for x in g['instructions'] if x['opcode']=='jmp');word=bits(angle)&0xffffffff;t=run(g,g['instructions'][0]['address'],[jump],constant_memory,{'%xmm0':word,'%rdi':MSG});args=[t['registers'][k] for k in ['rdi','rsi','rdx']];m=message_memory(64)
for x in graphs['MSG_WriteBits']['instructions']:
if x['opcode']=='addl' and '(%rip)' in x['operands']:put_word(m,global_address(x),0)
m,requests,error,child=writer(graphs,m,args[2],args[1]);size=get_word(m,MSG+28);write_bit=get_word(m,MSG+36)
if requests or error:raise ValueError('angle OOB helper')
decoded=None;read_count=None;read_bit=None
if name=='MSG_WriteAngle16':
m.update(constant_memory);put_word(m,MSG+32,0);put_word(m,MSG+36,7)
m,raw,read_error,rt=reader(graphs,m,16);decoded,tail,_=read_angle(graphs['MSG_ReadAngle16'],m,raw);child+=rt+tail
if read_error:raise ValueError('angle16 OOB reader')
read_count=get_word(m,MSG+32);read_bit=get_word(m,MSG+36)
writes.append({'function':name,'input_word':word,'arguments':args,'scaled_word':t['registers']['xmm0'],'payload_hex':bytes(m[0xa0000+i] for i in range(size)).hex(),'size':size,'bit':write_bit,'angle16_read_word':decoded,'read_count':read_count,'read_bit':read_bit,**trace_fields({'trace':t['trace']+child},False)})
reads=[]
for word,size,count in itertools.product([0,1,32767,32768,65535,0x12348001,0xffffffff],[0,2,0xffffffff],[0,2,3,0x80000000,0xffffffff]):
m=dict(constant_memory);put_word(m,MSG+28,size);put_word(m,MSG+32,count);result,trace,omitted=read_angle(graphs['MSG_ReadAngle16'],m,word);reads.append({'supplied_word':word,'size_word':size,'post_count_word':count,'result_word':result,'omitted_frame_instructions':omitted,**trace_fields({'trace':trace},False)})
out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,'file_backed_constants':constants,'float_oob':floats,'float_tails':tails,'angle_writes':writes,'angle16_tails':reads,'read_angle_symbol_present':'MSG_ReadAngle' in fs}
write_json(Path('docs/research/message-numeric.json'),out)
print('Recovered raw float transport and bounded finite angle conversions')
if __name__=='__main__':main()