File size: 7,621 Bytes
c057512 | 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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | #!/usr/bin/env python3
"""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
# Result of team-mode helper 33715 is an explicit fixture input here.
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 []})
# Retain direct immediate strings only at actual Q_stricmp setup sites.
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()
|