| |
| """Recover bounded chat assembly/recipient gates and native command metadata.""" |
| import gzip |
| import json |
| from pathlib import Path |
| import re |
| import struct |
|
|
| from analyze_deferred_auth import partition_function, trace_slice |
| from analyze_engine_control import control_flow |
| from analyze_event_queues import symbol_record |
| from analyze_frame_timing import read_address |
| from analyze_game_cvars import vm_string |
| from analyze_native import elf |
| from analyze_qvm import decode, write_json |
| from analyze_vm_extensions import native_functions, verified |
|
|
|
|
| def native_string(data, sections, address, limit=512): |
| result=bytearray() |
| for at in range(address,address+limit): |
| b,_=read_address(data,sections,at,1) |
| if b==b'\0':return result.decode('ascii') |
| result.extend(b) |
| raise ValueError('unterminated native string') |
|
|
|
|
| def main(): |
| q=json.loads(Path('docs/research/qvm/qagame.json').read_text()) |
| d=verified(Path(q['source']['member']['output']).read_bytes(),q['sha256']);h,ins=decode(d) |
| registry=json.loads(Path('docs/research/game-cvars.json').read_text()) |
| if registry['sha256']!=q['sha256']:raise ValueError('registry provenance mismatch') |
| row=next(r for r in registry['registry']['rows'] |
| if r['name']=='g_allowChat') |
| if row['mirror_address']+12!=34077172:raise ValueError('chat cvar identity changed') |
| joins=[] |
| for used,length,index,argc in [(0,0,0,1),(0,5,0,2),(1021,1,0,2),(1022,0,0,2),(1022,1,0,2),(0,1023,0,1)]: |
| accepted=used+length<1023 |
| t=trace_slice(ins,180896,180950 if accepted else 180955,{}, {20:used,28:length,24:index,1056:argc}) |
| after=t['memory'][0x40000000+20] |
| end=trace_slice(ins,180955,180962,{}, {20:after},capture_stack=True) |
| joins.append({'used':used,'token_length':length,'index':index,'argc':argc,'accepted':accepted, |
| 'trace':t,'terminator_trace':end}) |
| gates=[] |
| for allow in [0,1,2,3]: |
| for mode in [0,1,2,4]: |
| accepted=allow!=0 and (mode!=0 or allow>=2) |
| t=trace_slice(ins,186750,186764 if accepted else 187379,{34077172:allow},{124:mode}) |
| gates.append({'allow_chat':allow,'mode':mode,'accepted':accepted,'trace':t}) |
| sender,recipient,srecord,rrecord,base=0x10000,0x20000,0x30000,0x40000,0x30000-2*6336 |
| mute=[] |
| for deadline in [999,1000,1001]: |
| mute.append({'deadline':deadline,'now':1000,'trace':trace_slice(ins,186737,186750 if deadline<=1000 else 187379, |
| {sender+520:srecord,srecord+3128:deadline,34992532:1000},{116:sender})}) |
| ignores=[] |
| for value in [0,1,-1]: |
| ignores.append({'sender_slot':2,'ignore_word':value,'trace':trace_slice(ins,186788,186813 if value==0 else 187379, |
| {sender+520:srecord,recipient+520:rrecord,34992496:base,rrecord+3780+2*4:value}, |
| {116:sender,120:recipient})}) |
| states=[] |
| for state in [0,1,2,3]: |
| states.append({'state_raw':state,'trace':trace_slice(ins,186851,186863 if state==2 else 187379, |
| {recipient+520:rrecord,rrecord+468:state},{120:recipient})}) |
| teams=[] |
| for team_sender,team_recipient,self_target,team_mode_result in [(1,1,False,1),(1,2,False,1),(1,1,True,0)]: |
| target=sender if self_target else recipient |
| tr=srecord if self_target else rrecord |
| accepted=team_sender==team_recipient or self_target |
| |
| memory={sender+520:srecord,target+520:tr,srecord+3096:team_sender,tr+3096:team_recipient} |
| teams.append({'same_object':self_target,'sender_team':team_sender,'recipient_team':team_recipient, |
| 'team_mode_helper_result':team_mode_result, |
| 'trace':trace_slice(ins,186888,186916 if accepted else 187379,memory,{76:team_mode_result,116:sender,120:target})}) |
| output={'source':q['source'],'sha256':q['sha256'],'allow_chat_registry':row,'join_traces':joins, |
| 'allow_chat_traces':gates,'mute_traces':mute,'ignore_traces':ignores,'recipient_state_traces':states, |
| 'team_comparison_traces':teams,'functions':{str(e):partition_function(ins,e,d,h) for e in [180858,186721,187381]}, |
| 'chat_tail_window':ins[189262:189342], |
| 'strings':{str(at):vm_string(d,h,at) for at in [77124,77117,77706,77676,77599,77828,77805,77728]}, |
| 'native':{},'limitations':'Static QVM/Linux ELF64 and bounded synthetic slices. Calls are recorded, ' |
| 'not entered; zero return values are placeholders. Join lengths/team-helper results are supplied inputs. ' |
| 'Native command tables and constants are file-backed metadata; graphs are syntactic, not execution. ' |
| 'Remaining recipient/deadchat/filter/format/queue/runtime paths are not a complete chat contract.'} |
| 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']) |
| metadata,symbols=elf(data);sections=metadata['sections'];a=r['artifacts']['disassembly.txt'] |
| fs=native_functions(verified(gzip.open(a['path'],'rb').read(),a['uncompressed_sha256']).decode()) |
| names=['SV_ExecuteClientCommand','Cmd_Args_Sanitize','SV_ClientCommand'] |
| graphs={name:control_flow(fs[name]) for name in names if name in fs} |
| table=symbol_record(symbols,'ucmds') |
| if table['bytes']%16 or table['bytes']>4096:raise ValueError('invalid ucmd table size') |
| raw,offset=read_address(data,sections,table['address'],table['bytes']) |
| entries=[] |
| for i in range(0,len(raw),16): |
| name,handler=struct.unpack_from('<2Q',raw,i) |
| entries.append({'index':i//16,'name_address':name,'handler_address':handler, |
| 'name':native_string(data,sections,name) if name else None, |
| 'handler_symbols':[s['name'] for s in symbols if s['address']==handler and not s['undefined']] if handler else []}) |
| |
| strings=[];g=graphs['SV_ExecuteClientCommand'];ii=g['instructions'] |
| for n,x in enumerate(ii): |
| if '<Q_stricmp>' not in x['operands']:continue |
| for prev in ii[max(0,n-4):n]: |
| match=re.match(r'\$0x([0-9a-f]+), %edi',prev['operands']) |
| if prev['opcode']=='movl' and match: |
| addr=int(match[1],16);strings.append({'call_address':x['address'],'address':addr,'text':native_string(data,sections,addr)}) |
| sanitizer=graphs['Cmd_Args_Sanitize'];addresses=[] |
| for x in sanitizer['instructions']: |
| match=re.search(r'#\s*0x([0-9a-f]+)',x['operands']) |
| if x['opcode']=='movzbl' and match: |
| at=int(match[1],16);b,fo=read_address(data,sections,at,1) |
| addresses.append({'instruction':x['address'],'address':at,'file_offset':fo,'byte':b[0]}) |
| output['native'][label]={'source':r['source'],'disassembly_sha256':a['uncompressed_sha256'],'graphs':graphs, |
| 'builtin_table':{**table,'file_offset':offset,'entries':entries},'limit_command_strings':strings, |
| 'sanitizer_constant_bytes':addresses,'argv_symbol':symbol_record(symbols,'cmd_argv')} |
| write_json(Path('docs/research/chat-paths.json'),output) |
| print('Recovered chat join/recipient gates and both Linux ELF64 command tables') |
|
|
|
|
| if __name__=='__main__':main() |
|
|
