| |
| """Static references and bounded synthetic-memory traces for deferred auth drops.""" |
| import gzip |
| import json |
| import math |
| import struct |
| from pathlib import Path |
| from analyze_qvm import decode, write_json |
| from analyze_vm_extensions import verified, native_functions |
| from analyze_auth_drop import ascii_literal |
| from analyze_native_dispatch import recover |
|
|
|
|
| def signed32(value): |
| value &= 0xffffffff |
| return value-0x100000000 if value & 0x80000000 else value |
|
|
|
|
| def trace_slice(instructions, start, stop, memory, locals_map, limit=256, byte_memory=None, capture_stack=False, |
| halt_on_leave=False, stop_targets=(), call_results=None, resume_at=None, local_base=0x40000000, skip_initial_stop=False, coherent_memory=False): |
| """Evaluate only a selected straight/bounded slice against synthetic memory. |
| |
| Record calls rather than entering callees. Reject unknown reads/opcodes and |
| control flow outside the selected slice. This is not a QVM runtime. |
| """ |
| if any(not start <= target <= stop for target in stop_targets): |
| raise ValueError('stop target outside bounded slice') |
| from qvm_byte_memory_support import read_bytes,read_byte,write_bytes |
| stack, args, calls, trace, byte_writes = [], {}, [], [], [] |
| memory = dict(memory) |
| bytes_map = {} if coherent_memory and byte_memory is None else None if byte_memory is None else dict(byte_memory) |
| for offset, value in locals_map.items(): |
| memory[local_base+offset] = value |
| pc = start if resume_at is None else resume_at |
| if not start <= pc <= stop: |
| raise ValueError('resume outside bounded slice') |
| while pc != stop and (pc not in stop_targets or skip_initial_stop and not trace): |
| if not start <= pc < stop or len(trace) >= limit: |
| raise ValueError('slice control flow escaped bound') |
| ins = instructions[pc] |
| op, value = ins['opcode'], ins['operand'] |
| trace.append(pc) |
| pc += 1 |
| if op == 'CONST': |
| stack.append(value) |
| elif op == 'LOCAL': |
| stack.append(local_base+value) |
| elif op == 'LOAD4': |
| address = stack.pop() |
| if coherent_memory: |
| stack.append(signed32(int.from_bytes(read_bytes(memory,bytes_map,address,4),'little'))) |
| else: |
| if address not in memory:raise ValueError('unknown synthetic memory read') |
| stack.append(memory[address]) |
| elif op == 'LOAD1': |
| address = stack.pop() |
| if coherent_memory:stack.append(read_byte(memory,bytes_map,address)) |
| else: |
| if bytes_map is None or address not in bytes_map:raise ValueError('unknown synthetic byte read') |
| stack.append(bytes_map[address] & 255) |
| elif op == 'SEX8': |
| value = stack.pop() & 255 |
| stack.append(value-256 if value & 128 else value) |
| elif op == 'NEGI': |
| stack.append(signed32(-stack.pop())) |
| elif op == 'STORE4': |
| value, address = stack.pop(), stack.pop() |
| if coherent_memory: |
| write_bytes(memory,bytes_map,{address+j:(value>>(8*j))&255 for j in range(4)}) |
| else:memory[address] = signed32(value) |
| if bytes_map is not None and not coherent_memory: |
| for byte in range(4): |
| bytes_map[address+byte] = (value >> (8*byte)) & 255 |
| elif op == 'STORE1': |
| value, address = stack.pop(), stack.pop() |
| byte_writes.append({'address': address, 'value': value & 255}) |
| if coherent_memory:write_bytes(memory,bytes_map,{address:value&255}) |
| elif bytes_map is not None:bytes_map[address] = value & 255 |
| elif op in ('ADD', 'SUB', 'MULI'): |
| right, left = stack.pop(), stack.pop() |
| stack.append(signed32(left+right if op == 'ADD' else left-right if op == 'SUB' else left*right)) |
| elif op in ('LSH', 'RSHI', 'RSHU'): |
| right, left = stack.pop(), stack.pop() |
| if not 0 <= right < 32: |
| raise ValueError('shift outside bounded evaluator domain') |
| shifted = left << right if op == 'LSH' else signed32(left) >> right if op == 'RSHI' else (left & 0xffffffff) >> right |
| stack.append(signed32(shifted)) |
| elif op in ('BAND', 'BOR', 'BXOR'): |
| right, left = stack.pop(), stack.pop() |
| stack.append(signed32(left & right if op == 'BAND' else left | right if op == 'BOR' else left ^ right)) |
| elif op in ('MULF', 'ADDF', 'SUBF', 'DIVF'): |
| right, left = stack.pop(), stack.pop() |
| a, b = [struct.unpack('<f', struct.pack('<I', v & 0xffffffff))[0] for v in (left, right)] |
| label = {'MULF':'multiply', 'ADDF':'add', 'SUBF':'subtract', 'DIVF':'divide'}[op] |
| if not math.isfinite(a) or not math.isfinite(b): |
| raise ValueError('nonfinite bounded float '+label) |
| if op == 'DIVF' and b == 0: |
| raise ValueError('zero bounded float divisor') |
| try: |
| result = struct.pack('<f', a*b if op == 'MULF' else a+b if op == 'ADDF' else a-b if op == 'SUBF' else a/b) |
| except OverflowError as exc: |
| raise ValueError('overflow bounded float '+label) from exc |
| if not math.isfinite(struct.unpack('<f', result)[0]): |
| raise ValueError('nonfinite bounded float result') |
| stack.append(signed32(struct.unpack('<I', result)[0])) |
| elif op == 'NEGF': |
| word = stack.pop() & 0xffffffff |
| if not math.isfinite(struct.unpack('<f', struct.pack('<I', word))[0]): |
| raise ValueError('nonfinite bounded float negate') |
| stack.append(signed32(word ^ 0x80000000)) |
| elif op == 'STORE2': |
| value, address = stack.pop(), stack.pop() |
| if bytes_map is None: |
| raise ValueError('halfword store requires byte memory') |
| for offset in range(2): |
| bytes_map[address+offset] = (value >> (8*offset)) & 255 |
| byte_writes.append({'address': address+offset, 'value': bytes_map[address+offset]}) |
| elif op == 'LOAD2': |
| address = stack.pop() |
| if bytes_map is None or any(address+offset not in bytes_map for offset in range(2)): |
| raise ValueError('unknown synthetic halfword read') |
| stack.append(bytes_map[address] | (bytes_map[address+1] << 8)) |
| elif op == 'SEX16': |
| value = stack.pop() & 65535 |
| stack.append(value-65536 if value & 32768 else value) |
| elif op in ('EQF', 'NEF', 'LTF', 'LEF', 'GTF', 'GEF'): |
| right, left = [struct.unpack('<f', struct.pack('<I', stack.pop() & 0xffffffff))[0] for _ in range(2)] |
| if not math.isfinite(left) or not math.isfinite(right): |
| raise ValueError('nonfinite bounded float comparison') |
| condition = {'EQF':left==right, 'NEF':left!=right, 'LTF':left<right, 'LEF':left<=right, 'GTF':left>right, 'GEF':left>=right}[op] |
| if condition: |
| pc = value |
| elif op == 'CVIF': |
| value = signed32(stack.pop()) |
| stack.append(signed32(struct.unpack('<I', struct.pack('<f', value))[0])) |
| elif op == 'CVFI': |
| value = struct.unpack('<f', struct.pack('<I', stack.pop() & 0xffffffff))[0] |
| if not math.isfinite(value) or not -(2**31) <= math.trunc(value) < 2**31: |
| raise ValueError('conversion outside bounded float domain') |
| stack.append(math.trunc(value)) |
| elif op == 'BLOCK_COPY': |
| source, dest = stack.pop(), stack.pop() |
| if coherent_memory: |
| if not 0<=value<=4096:raise ValueError('unsupported coherent copy length') |
| copied=read_bytes(memory,bytes_map,source,value) |
| write_bytes(memory,bytes_map,{dest+j:v for j,v in enumerate(copied)}) |
| continue |
| if value % 4 or source % 4 or dest % 4: |
| raise ValueError('unaligned bounded word copy') |
| copied = [] |
| for offset in range(0, value, 4): |
| if source+offset not in memory: |
| raise ValueError('unknown bounded copy read') |
| word = memory[source+offset] & 0xffffffff |
| if bytes_map is not None: |
| for byte in range(4): |
| if source+offset+byte in bytes_map: |
| word = (word & ~(255 << (8*byte))) | ((bytes_map[source+offset+byte] & 255) << (8*byte)) |
| copied.append(word) |
| for offset, word in enumerate(copied): |
| memory[dest+offset*4] = signed32(word) |
| if bytes_map is not None: |
| for byte in range(4): |
| bytes_map[dest+offset*4+byte] = (word >> (8*byte)) & 255 |
| elif op in ('DIVI', 'MODI'): |
| right, left = signed32(stack.pop()), signed32(stack.pop()) |
| if right == 0 or (left == -(2**31) and right == -1): |
| raise ValueError('division outside bounded evaluator domain') |
| quotient = abs(left) // abs(right) |
| quotient = -quotient if (left < 0) != (right < 0) else quotient |
| stack.append(quotient if op == 'DIVI' else left - quotient*right) |
| elif op in ('LTU', 'LEU', 'GTU', 'GEU'): |
| right, left = stack.pop() & 0xffffffff, stack.pop() & 0xffffffff |
| if {'LTU':left<right,'LEU':left<=right,'GTU':left>right,'GEU':left>=right}[op]: |
| pc = value |
| elif op in ('EQ', 'NE', 'LTI', 'LEI', 'GTI', 'GEI'): |
| right, left = signed32(stack.pop()), signed32(stack.pop()) |
| condition = {'EQ': left == right, 'NE': left != right, 'LTI': left < right, |
| 'LEI': left <= right, 'GTI': left > right, 'GEI': left >= right}[op] |
| if condition: |
| pc = value |
| elif op == 'ARG': |
| args[value] = stack.pop() |
| elif op == 'CALL': |
| calls.append({'instruction': ins['index'], 'target': stack.pop(), 'args': dict(args)}) |
| args = {} |
| stack.append(0 if call_results is None else call_results.get(ins['index'], 0)) |
| elif op == 'POP': |
| stack.pop() |
| elif op == 'JUMP': |
| pc = stack.pop() |
| elif op == 'LEAVE' and halt_on_leave: |
| break |
| else: |
| raise ValueError(f'unsupported slice opcode {op}') |
| result = {'trace': trace, 'calls': calls, 'memory': memory, 'byte_writes': byte_writes} |
| if bytes_map is not None: |
| result['byte_memory'] = bytes_map |
| if capture_stack: |
| result['stack'] = stack |
| if stop_targets: |
| result['stop'] = pc |
| return result |
|
|
|
|
| def partition_function(instructions, entry, data, header): |
| if instructions[entry]['opcode'] != 'ENTER': |
| raise ValueError('expected function entry') |
| end = next(i for i in range(entry+1, len(instructions)) if instructions[i]['opcode'] == 'ENTER') |
| body = instructions[entry:end] |
| literals = [] |
| for i in body: |
| if i['opcode'] == 'CONST' and instructions[i['index']+1]['opcode'] == 'ARG' and 0 <= i['operand'] < header['data_length']+header['literal_length']: |
| value = ascii_literal(data, header['data_offset']+i['operand'], header['data_offset']+header['data_length']+header['literal_length']) |
| if value: |
| literals.append({'instruction': i['index'], 'vm_address': i['operand'], 'text': value}) |
| return {'entry': entry, 'linear_end_exclusive': end, 'instructions': body, 'literal_argument_candidates': literals} |
|
|
|
|
| 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, instructions = decode(data) |
| entries = [2319, 5096, 69338, 15186, 165818, 173979, 174399] |
| functions = {str(e): partition_function(instructions, e, data, header) for e in entries} |
| refs = [] |
| for i in instructions: |
| if i['opcode'] == 'CONST' and i['operand'] == 3588: |
| entry = next(j for j in range(i['index'], -1, -1) if instructions[j]['opcode'] == 'ENTER') |
| refs.append({'instruction': i['index'], 'containing_entry': entry, |
| 'window': instructions[max(entry, i['index']-4):i['index']+9]}) |
| callers = {str(e): [i['index'] for i in instructions if i['opcode'] == 'CALL' and i['index'] and |
| instructions[i['index']-1]['opcode'] == 'CONST' and instructions[i['index']-1]['operand'] == e] |
| for e in entries} |
| synthetic = [] |
| record = 0x10000 |
| for state, pending, now in [(2, 1000, 1100), (2, 1000, 1101), (1, 1000, 1101), |
| (2, 0, 1101), (2, 1, 102), (2, 0x7ffffff0, -0x7fffff80)]: |
| model = trace_slice(instructions, 15878, 15922, |
| {record+468: state, record+3588: pending, 34992532: now}, {32: record, 28: 3}) |
| synthetic.append({'state': state, 'pending': pending, 'now': now, |
| 'signed_elapsed': signed32(now-pending), 'trace': model['trace'], 'calls': model['calls']}) |
| native = {} |
| 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()) |
| original = verified((Path('.cache/UrbanTerror43')/r['source']['path']).read_bytes(), r['source']['sha256']) |
| artifact = r['artifacts']['disassembly.txt'] |
| text = verified(gzip.open(artifact['path'], 'rb').read(), artifact['uncompressed_sha256']).decode() |
| dispatch = recover(original, text, 'SV_GameSystemCalls') |
| native[label] = {'source': r['source'], 'disassembly_sha256': artifact['uncompressed_sha256'], |
| 'strncpy_case': next(c for c in dispatch['cases'] if 102 in c['numbers_sharing_target']), |
| 'error_case': next(c for c in dispatch['cases'] if 1 in c['numbers_sharing_target']), |
| 'error_function': native_functions(text)['Com_Error']} |
| write_json(Path('docs/research/deferred-auth.json'), {'source': report['source'], 'sha256': report['sha256'], |
| 'functions': functions, 'direct_call_candidates': callers, 'field_constant_references': refs, |
| 'synthetic_slice_traces': synthetic, 'native': native, 'limitations': |
| 'Constant-offset references do not cover aliases or dynamically computed offsets. ' |
| 'Synthetic-memory traces evaluate selected QVM instructions without entering callees; ' |
| 'these are static boundary checks, not real gameplay/time/network fixtures or a full QVM runtime.'}) |
| print('Recovered eight field references, copy/error helpers and deferred timing boundary traces') |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
