File size: 5,674 Bytes
63e6366 | 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 | #!/usr/bin/env python3
"""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()
|