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