File size: 7,558 Bytes
55b95f3 | 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 | #!/usr/bin/env python3
"""Bounded keyed usercmd instruction compositions; no original code executes."""
import gzip
import hashlib
import json
from pathlib import Path
from analyze_native import elf
from analyze_bsp_loading import read_address
from analyze_bsp_references import global_address, trace_fields
from analyze_collision_composition import get_word, put_word
from analyze_engine_control import control_flow
from analyze_message_byte_codec import MSG, DATA, install_tree
from analyze_message_numeric import frame_graph
from analyze_message_widths import reader, run, writer
from analyze_message_wrappers import WRAPPER_STACK
from analyze_qvm import write_json
from analyze_vm_extensions import native_functions, verified
BASE=0xd0000
NEW=0xe0000
DEST=0xf0000
FIELDS=[(4,4,16),(8,4,16),(12,4,16),(21,1,8),(22,1,8),(23,1,8),(16,4,16),(20,1,8)]
def compose(graphs,m,key,reading=False):
name='MSG_ReadDeltaUsercmdKey' if reading else 'MSG_WriteDeltaUsercmdKey'
g,omitted=frame_graph(graphs[name]);rows=g['instructions'];calls={x['address']:i for i,x in enumerate(rows) if x['opcode']=='callq'};tails={x['address'] for x in rows if x['opcode']=='jmp' and '<MSG_WriteBits>' in x['operands']};ends={x['address'] for x in rows if x['opcode']=='retq'}
regs={'%rdi':MSG,'%rsi':key,'%rdx':BASE,'%rcx':DEST if reading else NEW,'%rsp':WRAPPER_STACK};pc=rows[0]['address'];requests=[];trace=[]
while True:
t=run(g,pc,list(calls)+list(tails)+list(ends),m,regs);m=t['memory'];regs=t['registers'];trace+=t['trace']
if t['stop'] in ends:break
request=[regs[k] for k in (['rdi','rsi'] if reading else ['rdi','rsi','rdx'])];requests.append(request)
if request[0]!=MSG:raise ValueError('usercmd pointer')
if reading:
m,value,error,child=reader(graphs,m,request[1]);regs['rax']=value
else:m,_,error,child=writer(graphs,m,request[2],request[1])
trace+=child
if error:raise ValueError('usercmd child error')
if t['stop'] in tails:break
pc=rows[calls[t['stop']]+1]['address']
if len(requests)>19:raise ValueError('usercmd request bound')
return m,requests,trace,omitted
def fixtures():
baseline=bytearray.fromhex('640000003412000078560000bc9a0000420000000780ff01')
for delta in [0,1,255,256,-1]:
for mutation in [-1,*range(8),8]:
new=bytearray(baseline);new[:4]=((100+delta)&0xffffffff).to_bytes(4,'little')
for i,(offset,size,_) in enumerate(FIELDS):
if mutation in [i,8]:new[offset:offset+size]=((int.from_bytes(new[offset:offset+size],'little')+1)&((1<<(8*size))-1)).to_bytes(size,'little')
for key in [0,0x10100]:yield delta,mutation,key,0,bytes(baseline),bytes(new)
b=bytearray(baseline);b[:4]=(0xfffffffe).to_bytes(4,'little');n=bytearray(b);n[:4]=(1).to_bytes(4,'little');n[4:8]=(0xffffffff).to_bytes(4,'little')
for key in [0,0xffffffff]:yield 'wrap',0,key,7,bytes(b),bytes(n)
def main():
cp=Path('docs/research/huffman-codebook.json');ip=Path('docs/research/huffman-initialization.json');code=json.loads(cp.read_text());initial=json.loads(ip.read_text())
out={'native':{},'codebook_report_sha256':hashlib.sha256(cp.read_bytes()).hexdigest(),'initialization_report_sha256':hashlib.sha256(ip.read_bytes()).hexdigest(),'field_layout':FIELDS,'limitations':'Synthetic Linux ELF64 instruction compositions only; no original process/native callee executes. Compressed trees MODEL-DERIVED, not full native training/captured packets. PUSH/POP and stack save/restores omitted; disjoint frames and explicit caller-register restoration. Ample zero-initialized buffers; selected timestamps/fields/keys only. Record field semantic names inferred separately. Negative timestamp and high key effects are synthetic reachability probes, not legal game traffic. No overflow/EOF/alias/full caller/runtime/platform conformance.'}
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,symbols=elf(data);a=r['artifacts']['disassembly.txt'];fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode());names=['MSG_WriteDeltaUsercmdKey','MSG_ReadDeltaUsercmdKey','MSG_WriteBits','MSG_ReadBits','Huff_putBit','Huff_getBit','Huff_offsetTransmit','Huff_offsetReceive','send'];graphs={n:control_flow(fs[n]) for n in names}
for dep in [code,initial]:
if dep['native'][label]['source']!=r['source'] or dep['native'][label]['disassembly_sha256']!=a['uncompressed_sha256']:raise ValueError('dependency provenance')
mask=next(s for s in symbols if s['name']=='kbitmask');raw,offset=read_address(data,meta['sections'],mask['address'],128);template={mask['address']+i:v for i,v in enumerate(raw)};base=initial['native'][label]['entry_request']['huff_init_pointer'];state=code['model_derived_seed_training']['final_state'];install_tree(template,state,base);root=install_tree(template,state,base+0xc830)
for x in graphs['MSG_ReadBits']['instructions']:
if x['opcode']=='movq' and '(%rip)' in x['operands']:put_word(template,global_address(x),root,8)
g=graphs['MSG_WriteBits'];old=global_address(next(x for x in g['instructions'] if x['opcode']=='addl' and '(%rip)' in x['operands']));over=global_address(next(x for x in g['instructions'] if x['opcode']=='addl' and '$0x1,' in x['operands'] and '(%rip)' in x['operands']));streams=[]
for delta,mutation,key,bit,b,n in fixtures():
m=dict(template)
for i in range(256):m[DATA+i]=0
for ptr,record in [(BASE,b),(NEW,n),(DEST,bytes([0xa5]*24))]:
m[ptr-1]=m[ptr+24]=0x7b
for i,v in enumerate(record):m[ptr+i]=v
m[DATA-1]=m[DATA+256]=0x7b
for o,v in [(4,0),(8,0),(16,DATA),(24,256),(28,bit//8+1),(32,0),(36,bit)]:put_word(m,MSG+o,v,8 if o==16 else 4)
put_word(m,old,0);put_word(m,over,0);m,wq,trace,wo=compose(graphs,m,key);wb=get_word(m,MSG+36);size=get_word(m,MSG+28);payload=bytes(m[DATA+i] for i in range(size)).hex();account=get_word(m,old);counter=get_word(m,over)
put_word(m,MSG+36,bit);put_word(m,MSG+32,0);m,rq,child,ro=compose(graphs,m,key,True);trace+=child
streams.append({'delta_fixture':delta,'mutation':mutation,'key_word':key,'initial_bit':bit,'baseline_hex':b.hex(),'new_hex':n.hex(),'decoded_hex':bytes(m[DEST+i] for i in range(24)).hex(),'write_requests':wq,'read_requests':rq,'payload_hex':payload,'write_bit':wb,'write_size':size,'read_bit':get_word(m,MSG+36),'read_count':get_word(m,MSG+32),'oldsize_word':account,'range_counter':counter,'overflow_flag':get_word(m,MSG+4),'canaries':[m[p] for p in [DATA-1,DATA+256,BASE-1,BASE+24,NEW-1,NEW+24,DEST-1,DEST+24]],'inputs_unchanged':bytes(m[BASE+i] for i in range(24))==b and bytes(m[NEW+i] for i in range(24))==n,'writer_omitted':wo,'reader_omitted':ro,**trace_fields({'trace':trace},False)})
out['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs,'mask_symbol':mask,'mask_file_offset':offset,'mask8':int.from_bytes(raw[32:36],'little'),'mask16':int.from_bytes(raw[64:68],'little'),'streams':streams}
write_json(Path('docs/research/message-usercmd.json'),out);print('Recovered keyed usercmd timestamp/field boundaries and modeled-tree compositions')
if __name__=='__main__':main()
|