| |
| """Selected native control-flow evidence for simulation timing and errors.""" |
| import gzip |
| import json |
| from pathlib import Path |
| import re |
| from analyze_qvm import decode, write_json |
| from analyze_vm_extensions import verified, native_functions, command_callers |
| from analyze_deferred_auth import partition_function, trace_slice |
|
|
|
|
| def control_flow(function): |
| instructions = [] |
| for line in function['lines']: |
| m = re.match(r'\s*([0-9a-f]+):[^\t]*\t(\w+)\s*(.*)', line) |
| if m: |
| instructions.append({'address': int(m[1], 16), 'opcode': m[2], 'operands': m[3], 'line': line.strip()}) |
| if not instructions or instructions[0]['address'] != function['address']: |
| raise ValueError('missing function entry instruction') |
| addresses = {i['address'] for i in instructions} |
| if len(addresses) != len(instructions): |
| raise ValueError('duplicate instruction addresses') |
| edges, calls = [], [] |
| for index, ins in enumerate(instructions): |
| op, operands, address = ins['opcode'], ins['operands'], ins['address'] |
| next_address = instructions[index+1]['address'] if index+1 < len(instructions) else None |
| target = re.match(r'0x([0-9a-f]+)', operands) |
| target = int(target[1], 16) if target else None |
| name = re.search(r'<([^>]+)>', operands) |
| if op.startswith('call'): |
| calls.append({'address': address, 'target': target, 'label': name[1] if name else None}) |
| if op.startswith(('j', 'loop')): |
| edges.append({'from': address, 'to': target, 'kind': 'unconditional' if op in ('jmp', 'jmpq') else 'conditional_taken', |
| 'target_scope': 'unresolved_indirect' if target is None else 'local' if target in addresses else 'external'}) |
| if op in ('jmp', 'jmpq'): |
| continue |
| if op.startswith('ret'): |
| edges.append({'from': address, 'to': None, 'kind': 'return', 'target_scope': 'external'}) |
| continue |
| if next_address is not None: |
| edges.append({'from': address, 'to': next_address, |
| 'kind': 'possible_call_continuation' if op.startswith('call') else 'fallthrough', 'target_scope': 'local'}) |
| return {'entry': function['address'], 'instructions': instructions, 'edges': edges, 'calls': calls, |
| 'limitations': 'Syntactic instruction graph, not a reachability proof. Computed jumps remain unresolved; ' |
| 'calls retain possible continuation even for known nonreturning external services.'} |
|
|
|
|
| def windows(function, patterns, before=4, after=8): |
| ranges = set() |
| lines = function['lines'] |
| for i, line in enumerate(lines): |
| if any(re.search(pattern, line) for pattern in patterns): |
| ranges.update(range(max(0, i-before), min(len(lines), i+after))) |
| return [lines[i].strip() for i in sorted(ranges)] |
|
|
|
|
| def main(): |
| output = {'native': {}, 'limitations': 'Static Linux ELF64 evidence. No original game process is launched. ' |
| 'Instruction windows and control-flow edges are evidence rather than complete decompilation or platform parity.'} |
| names = ['Com_Error', 'Com_Frame', 'SV_Frame', 'Sys_Milliseconds', 'Sys_Error'] |
| for label in ('linux--Quake3-UrT-Ded.x86_64', 'linux--Quake3-UrT.x86_64'): |
| report = json.loads(Path(f'docs/research/native/{label}.json').read_text()) |
| verified((Path('.cache/UrbanTerror43')/report['source']['path']).read_bytes(), report['source']['sha256']) |
| artifact = report['artifacts']['disassembly.txt'] |
| text = verified(gzip.open(artifact['path'], 'rb').read(), artifact['uncompressed_sha256']).decode() |
| functions = native_functions(text) |
| output['native'][label] = {'source': report['source'], 'disassembly_sha256': artifact['uncompressed_sha256'], |
| 'graphs': {name: control_flow(functions[name]) for name in names}, |
| 'frame_command_candidates': [r for r in command_callers({'SV_Frame': functions['SV_Frame']}) if r['command_candidate'] == 8], |
| 'clock_windows': windows(functions['SV_Frame'], [r'<sv_fps>', r'\$0x3e8', r'<VM_Call>', r'\$0x70000000']), |
| 'error_windows': windows(functions['Com_Error'], [r'\$0x63', r'<errorCount', r'<com_buildScript>', r'<com_errorEntered>', r'longjmp', r'<Sys_Error>'])} |
| report = json.loads(Path('docs/research/qvm/qagame.json').read_text()) |
| data = verified(Path(report['source']['member']['output']).read_bytes(), report['sha256']) |
| header, instructions = decode(data) |
| helper = partition_function(instructions, 165352, data, header) |
| synthetic = [] |
| record = 0x10000 |
| for enabled, pending, announced in [(1, 0, 0), (1, 0, 1), (1, 1, 0), (1, 1000, 0), (0, 0, 0)]: |
| model = trace_slice(instructions, 174979, 175027, |
| {35017616: enabled, record+3588: pending, record+3200: announced, 34992532: 1234}, {24: record}) |
| synthetic.append({'positive_auth_gate': enabled, 'pending': pending, 'announcement_flag': announced, |
| 'result_flag': model['memory'][record+3200], 'calls': model['calls'], 'trace': model['trace']}) |
| output['qvm'] = {'source': report['source'], 'sha256': report['sha256'], 'announcement_helper': helper, |
| 'begin_slice': instructions[174979:175027], 'synthetic_gate_traces': synthetic} |
| write_json(Path('docs/research/engine-control.json'), output) |
| print('Recovered ten native instruction graphs, clock/error windows and auth announcement gate') |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
