| |
| """Static Linux frame evidence and a bounded Com_ModifyMsec instruction evaluator.""" |
| import gzip |
| import json |
| import math |
| from pathlib import Path |
| import re |
| import struct |
|
|
| from analyze_engine_control import control_flow, windows |
| from analyze_native import elf |
| from analyze_qvm import write_json |
| from analyze_vm_extensions import native_functions, verified |
|
|
|
|
| def read_address(data, sections, address, size): |
| if size < 0: |
| raise ValueError('negative read size') |
| matches = [s for s in sections if not s['zero_fill'] and |
| s['address'] <= address and address + size <= s['address'] + s['bytes']] |
| if len(matches) != 1: |
| raise ValueError('address is not uniquely file-backed') |
| offset = matches[0]['file_offset'] + address - matches[0]['address'] |
| if offset < 0 or offset + size > len(data): |
| raise ValueError('section read outside file') |
| return data[offset:offset+size], offset |
|
|
|
|
| def float_constants(data, sections, graph): |
| result = [] |
| for ins in graph['instructions']: |
| if ins['opcode'] != 'comiss' or '(%rip)' not in ins['operands']: |
| continue |
| match = re.search(r'#\s*0x([0-9a-f]+)', ins['operands']) |
| if not match: |
| raise ValueError('missing RIP-relative constant address') |
| address = int(match[1], 16) |
| raw, offset = read_address(data, sections, address, 4) |
| result.append({'instruction': ins['address'], 'address': address, |
| 'file_offset': offset, 'hex': raw.hex(), 'float32': struct.unpack('<f', raw)[0]}) |
| return result |
|
|
|
|
| def event_table(data, sections, graph): |
| jumps = [i for i in graph['instructions'] if i['opcode'] == 'jmpq' and |
| re.fullmatch(r'\*0x[0-9a-f]+\(,%rax,8\)', i['operands'])] |
| if len(jumps) != 1: |
| raise ValueError('unexpected event switch shape') |
| ins = jumps[0] |
| base = int(re.search(r'0x([0-9a-f]+)', ins['operands'])[1], 16) |
| |
| if not any(i['opcode'] == 'cmpl' and i['operands'].startswith('$0x6, %edx') |
| for i in graph['instructions']): |
| raise ValueError('missing event switch upper-bound evidence') |
| raw, offset = read_address(data, sections, base, 7*8) |
| targets = struct.unpack('<7Q', raw) |
| by_address = {i['address']: k for k, i in enumerate(graph['instructions'])} |
| if any(t not in by_address for t in targets): |
| raise ValueError('event table target outside function') |
| return {'instruction': ins['address'], 'address': base, 'file_offset': offset, |
| 'raw_hex': raw.hex(), 'entries': [{'event_type': n, 'target': t, |
| 'entry_window': graph['instructions'][by_address[t]:by_address[t]+16]} |
| for n, t in enumerate(targets)], |
| 'limitations': 'Type zero is handled before the switch. Entry windows are linear evidence, not case reachability.'} |
|
|
|
|
| def pause_assignments(data, sections, graph): |
| result = [] |
| for k, ins in enumerate(graph['instructions']): |
| if ins['opcode'] != 'callq' or '<Cvar_Set>' not in ins['operands']: |
| continue |
| arguments = {} |
| for previous in graph['instructions'][max(0, k-3):k]: |
| match = re.match(r'\$0x([0-9a-f]+), %(edi|esi)(?:\s|$)', previous['operands']) |
| if previous['opcode'] != 'movl' or not match: |
| continue |
| address = int(match[1], 16) |
| payload = bytearray() |
| for at in range(address, address+256): |
| raw, _ = read_address(data, sections, at, 1) |
| if raw == b'\0': |
| break |
| payload.extend(raw) |
| else: |
| raise ValueError('unterminated cvar argument') |
| arguments[match[2]] = {'address': address, 'text': payload.decode('ascii')} |
| if set(arguments) != {'edi', 'esi'}: |
| raise ValueError('missing direct pause cvar arguments') |
| result.append({'call': ins['address'], 'arguments': arguments}) |
| return result |
|
|
|
|
| def f32(value): |
| try: |
| return struct.unpack('<f', struct.pack('<f', value))[0] |
| except OverflowError as exc: |
| raise ValueError('float operation outside evaluator domain') from exc |
|
|
|
|
| def evaluate_modify(graph, constants, values): |
| """Interpret only the observed scalar subset; no callees or game code execute. |
| |
| Input domain: signed-int32 inputs and finite, representable float operations. |
| Pointer registers are symbolic cvar handles. Calls only record a diagnostic. |
| Unsupported operands, opcodes and arithmetic domains fail closed. |
| """ |
| required = {'elapsed', 'com_fixedtime', 'com_timescale', 'com_cameraMode', |
| 'com_dedicated', 'com_sv_running'} |
| if set(values) != required: |
| raise ValueError('unexpected timing input fields') |
| for key in required - {'com_timescale'}: |
| if not isinstance(values[key], int) or not -(2**31) <= values[key] < 2**31: |
| raise ValueError('input outside signed-int32 domain') |
| scale = f32(values['com_timescale']) |
| if not math.isfinite(scale): |
| raise ValueError('nonfinite timescale unsupported') |
| values = dict(values, com_timescale=scale) |
| instructions = graph['instructions'] |
| indexes = {i['address']: n for n, i in enumerate(instructions)} |
| const = {c['instruction']: c['float32'] for c in constants} |
| regs = {'%edi': values['elapsed']} |
| trace, calls = [], [] |
| equal = less = greater = False |
| pc = 0 |
|
|
| def operand(text): |
| if text.startswith('$'): |
| value = int(text[1:], 0) |
| return value - 2**32 if 2**31 <= value < 2**32 else value |
| if text in regs: |
| return regs[text] |
| match = re.fullmatch(r'0x(2c|30)\((%\w+)\)', text) |
| if match: |
| kind, key = regs[match[2]] |
| if kind != 'cvar' or ((match[1] == '2c') != (key == 'com_timescale')): |
| raise ValueError('unexpected cvar field access') |
| return values[key] |
| raise ValueError('unsupported operand: ' + text) |
|
|
| for _ in range(1000): |
| ins = instructions[pc] |
| trace.append(ins['address']) |
| op = ins['opcode'] |
| args = ins['operands'].split('#')[0].strip() |
| fields = [x.strip() for x in args.split(',')] |
| next_pc = pc + 1 |
| if op in ('pushq', 'popq', 'nop', 'nopw', 'nopl'): |
| pass |
| elif op == 'movq' and '(%rip)' in args: |
| match = re.search(r'<(com_\w+)>', ins['operands']) |
| if not match or match[1] not in required: |
| raise ValueError('unexpected cvar pointer') |
| regs[fields[1]] = ('cvar', match[1]) |
| elif op in ('movl', 'movss'): |
| regs[fields[1]] = operand(fields[0]) |
| elif op == 'xorps' and fields[0] == fields[1]: |
| regs[fields[1]] = 0.0 |
| elif op == 'cvtsi2ss': |
| regs[fields[1]] = f32(operand(fields[0])) |
| elif op == 'mulss': |
| regs[fields[1]] = f32(operand(fields[0]) * operand(fields[1])) |
| elif op == 'cvttss2si': |
| number = operand(fields[0]) |
| if not math.isfinite(number) or not -(2**31) <= math.trunc(number) < 2**31: |
| raise ValueError('float conversion outside evaluator domain') |
| regs[fields[1]] = math.trunc(number) |
| elif op in ('cmpl', 'comiss', 'testl'): |
| a = const[ins['address']] if ins['address'] in const else operand(fields[0]) |
| b = operand(fields[1]) |
| if op == 'testl': |
| if fields[0] != fields[1]: |
| raise ValueError('unsupported bit test') |
| a, b = 0, b |
| equal, less, greater = b == a, b < a, b > a |
| elif op in ('cmovlel', 'cmovnel'): |
| condition = (less or equal) if op == 'cmovlel' else not equal |
| if condition: |
| regs[fields[1]] = operand(fields[0]) |
| elif op in ('jne', 'je', 'jle', 'jg', 'jmp'): |
| take = {'jne': not equal, 'je': equal, 'jle': less or equal, |
| 'jg': greater, 'jmp': True}[op] |
| if take: |
| next_pc = indexes[int(fields[0].split()[0], 16)] |
| elif op == 'xorl' and fields[0] == fields[1]: |
| regs[fields[1]] = 0 |
| equal, less, greater = True, False, False |
| elif op == 'callq' and '<Com_Printf>' in ins['operands']: |
| calls.append({'address': ins['address'], 'service': 'Com_Printf', |
| 'msec_argument': regs['%esi']}) |
| |
| for register in ('%eax', '%ecx', '%edx', '%esi', '%edi', '%r8', '%r9', '%r10', '%r11'): |
| regs.pop(register, None) |
| elif op == 'retq': |
| return {'result': regs['%eax'], 'trace': trace, 'calls': calls} |
| else: |
| raise ValueError('unsupported instruction: ' + ins['line']) |
| pc = next_pc |
| raise ValueError('instruction budget exceeded') |
|
|
|
|
| def timing_cases(): |
| defaults = {'elapsed': 16, 'com_fixedtime': 0, 'com_timescale': 1.0, |
| 'com_cameraMode': 0, 'com_dedicated': 0, 'com_sv_running': 1} |
| cases = [dict(defaults)] |
| for field, samples in [('elapsed', [0, -1, 200, 201, 500, 501, 5000, 5001]), |
| ('com_fixedtime', [-5, 1, 250, 6000]), |
| ('com_timescale', [0.0, -1.0, 0.01, 0.5, 1.5, 2.0]), |
| ('com_cameraMode', [1])]: |
| cases.extend(dict(defaults, **{field: value}) for value in samples) |
| cases.extend(dict(defaults, elapsed=elapsed, com_dedicated=ded, com_sv_running=running) |
| for elapsed in [500, 501, 5001] for ded, running in [(1, 1), (1, 0), (0, 0)]) |
| cases.extend([dict(defaults, com_fixedtime=-5, com_timescale=0.0), |
| dict(defaults, com_cameraMode=1, com_timescale=0.0), |
| dict(defaults, com_fixedtime=17, com_timescale=2.0)]) |
| return cases |
|
|
|
|
| def main(): |
| output = {'native': {}, 'limitations': 'Static Linux ELF64 only. Original processes are not launched. ' |
| 'Synthetic native instruction evaluator is bounded to Com_ModifyMsec finite arithmetic; ' |
| 'it does not execute callees, model the event loop, or prove all-platform runtime parity.'} |
| names = ['Com_ModifyMsec', 'Com_GetRealEvent', 'Com_EventLoop', 'Sys_GetEvent', 'SV_CheckPaused'] |
| 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()) |
| data = verified((Path('.cache/UrbanTerror43') / report['source']['path']).read_bytes(), report['source']['sha256']) |
| artifact = report['artifacts']['disassembly.txt'] |
| functions = native_functions(verified(gzip.open(artifact['path'], 'rb').read(), artifact['uncompressed_sha256']).decode()) |
| metadata, _ = elf(data) |
| graphs = {name: control_flow(functions[name]) for name in names} |
| constants = float_constants(data, metadata['sections'], graphs['Com_ModifyMsec']) |
| frame = control_flow(functions['Com_Frame']) |
| output['native'][label] = {'source': report['source'], 'disassembly_sha256': artifact['uncompressed_sha256'], |
| 'graphs': graphs, 'modify_float_constants': constants, |
| 'event_switch': event_table(data, metadata['sections'], graphs['Com_EventLoop']), |
| 'pause_assignments': pause_assignments(data, metadata['sections'], graphs['SV_CheckPaused']), |
| 'frame_float_constants': float_constants(data, metadata['sections'], frame), |
| 'frame_windows': windows(functions['Com_Frame'], [r'<lastTime\.', r'<com_fixedtime>', r'<com_timescale>', r'<com_cameraMode>', r'<com_maxfps>', r'<SV_Frame>', r'<CL_Frame>', r'<Com_EventLoop>', r'\$0x1f4', r'\$0x1388'], 3, 12), |
| 'server_pause_windows': windows(functions['SV_Frame'], [r'<cl_paused>', r'<sv_paused>', r'\$0x24ee4', r'\$0x2df40'], 4, 12), |
| 'synthetic_modify_cases': [{'input': case, **evaluate_modify(graphs['Com_ModifyMsec'], constants, case)} for case in timing_cases()]} |
| write_json(Path('docs/research/frame-timing.json'), output) |
| print(f'Recovered event/timing/pause graphs and {2*len(timing_cases())} bounded native instruction traces') |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
