File size: 6,141 Bytes
425d241 | 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 | #!/usr/bin/env python3
"""Recover the qagame cvar registry and authentication announcement callers."""
import gzip
import json
from pathlib import Path
import struct
from analyze_deferred_auth import partition_function, trace_slice
from analyze_engine_control import control_flow
from analyze_qvm import decode, write_json
from analyze_vm_extensions import native_functions, verified
def vm_string(data, header, address):
length = header['data_length'] + header['literal_length']
if not 0 <= address < length:
raise ValueError('string outside initialized VM memory')
start, stop = header['data_offset']+address, header['data_offset']+length
if stop > len(data):
raise ValueError('truncated initialized VM memory')
end = data.find(b'\0', start, stop)
if end == -1:
raise ValueError('unterminated VM string')
return data[start:end].decode('ascii')
def registry(data, header, base, count, stride=28):
if stride != 28 or not 0 < count <= 4096 or base < 0 or base+count*stride > header['data_length']:
raise ValueError('invalid cvar table bounds')
rows = []
for index in range(count):
at = base+index*stride
offset = header['data_offset']+at
if offset+stride > len(data):
raise ValueError('truncated cvar table')
pointer, name, default, flags, cache, word20, word24 = struct.unpack_from('<7I', data, offset)
if pointer and not 0 <= pointer <= header['declared_memory_bytes']-272:
raise ValueError('cvar mirror outside declared memory')
text = vm_string(data, header, name)
if not text:
raise ValueError('empty cvar name')
rows.append({'index': index, 'vm_address': at, 'file_offset': offset, 'mirror_address': pointer,
'name_address': name, 'name': text, 'default_address': default,
'default': vm_string(data, header, default), 'flags_raw': flags,
'modification_cache_initial': cache, 'word20_initial': word20, 'word24_initial': word24})
return rows
def main():
report = json.loads(Path('docs/research/qvm/qagame.json').read_text())
data = verified(Path(report['source']['member']['output']).read_bytes(), report['sha256'])
header, ins = decode(data)
# Concrete identities are checked against the supplied registration/update loops.
for at, value in [(2685, 4), (2783, 28), (2788, 3532), (2958, 4), (3843, 28), (3848, 3532)]:
if ins[at]['opcode'] != 'CONST' or ins[at]['operand'] != value:
raise ValueError('cvar loop identity changed')
count = struct.unpack_from('<I', data, header['data_offset']+3532)[0]
rows = registry(data, header, 4, count)
matches = [r for r in rows if r['name'] == 'auth_verbosity']
if len(matches) != 1 or matches[0]['mirror_address']+12 != 35002972:
raise ValueError('announcement cvar identity mismatch')
memory = {at: struct.unpack_from('<I', data, header['data_offset']+at)[0]
for at in range(4, 3532, 4)}
registration_traces = []
for row in rows:
trace = trace_slice(ins, 2689, 2714, memory, {24: row['vm_address']})
registration_traces.append({'index': row['index'], 'calls': trace['calls'], 'trace': trace['trace']})
callers, globals_refs, entry = [], [], None
for i in ins:
if i['opcode'] == 'ENTER':
entry = i['index']
if i['opcode'] == 'CONST' and i['operand'] == 165352 and ins[i['index']+1]['opcode'] == 'CALL':
callers.append({'instruction': i['index']+1, 'function_entry': entry,
'argument_window': ins[max(entry, i['index']-16):i['index']+3]})
if i['opcode'] == 'CONST' and i['operand'] == 35002972:
globals_refs.append({'instruction': i['index'], 'function_entry': entry,
'window': ins[max(entry, i['index']-4):i['index']+7]})
record = 0x10000
tail = trace_slice(ins, 1370, 1381, {}, {20: record, 40: 0x30000})
response_traces = []
for verbosity in [-1, 0, 1, 2, 3]:
trace = trace_slice(ins, 166732, 166746, {35002972: verbosity}, {36: record})
response_traces.append({'verbosity': verbosity, 'calls': trace['calls'], 'trace': trace['trace']})
output = {'source': report['source'], 'sha256': report['sha256'],
'registry': {'base': 4, 'stride': 28, 'count_address': 3532, 'count': count, 'rows': rows},
'registration_traces': registration_traces, 'announcement_cvar': matches[0],
'announcement_direct_callers': callers, 'verbosity_literal_references': globals_refs,
'query_success_trace': tail, 'response_gate_traces': response_traces,
'qvm_functions': {str(e): partition_function(ins, e, data, header) for e in [1264, 2677, 2950]},
'native': {}, 'limitations': 'Static supplied qagame/Linux ELF64 plus bounded synthetic traces only. '
'Registry fields 20/24 retain raw values pending full behavioral recovery. Calls do not enter callees; '
'direct caller/reference scans exclude computed targets, bulk writes and aliases. Defaults are not effective runtime configuration.'}
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())
verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(), r['source']['sha256'])
artifact = r['artifacts']['disassembly.txt']
functions = native_functions(verified(gzip.open(artifact['path'], 'rb').read(), artifact['uncompressed_sha256']).decode())
output['native'][label] = {'source': r['source'], 'disassembly_sha256': artifact['uncompressed_sha256'],
'graphs': {name: control_flow(functions[name]) for name in ['Cvar_Register', 'Cvar_Update']}}
write_json(Path('docs/research/game-cvars.json'), output)
print(f'Recovered {count} registry entries, their registration argument traces and announcement callers')
if __name__ == '__main__':
main()
|