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