File size: 15,077 Bytes
529b124 f93de71 529b124 66f51f1 3ed15e9 529b124 e15838a 3ed15e9 529b124 3ed15e9 529b124 b9805ee 16c1385 529b124 3ed15e9 9ac939d 3ed15e9 9ac939d 529b124 3ed15e9 eaa66f5 529b124 3ed15e9 529b124 71a2cf7 f911441 71a2cf7 ab95057 b1c5e8e ab95057 ae4ba74 f93de71 ae4ba74 f93de71 48c7db8 ae4ba74 f93de71 ae4ba74 f93de71 48c7db8 f93de71 48c7db8 3bae649 f93de71 3ed15e9 f93de71 71a2cf7 f911441 71a2cf7 16c1385 9b92b16 16c1385 9b92b16 529b124 0113c21 529b124 66f51f1 529b124 9ac939d 66f51f1 e15838a 9ac939d 529b124 | 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 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 | #!/usr/bin/env python3
"""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)) # explicit supplied result or default placeholder
elif op == 'POP':
stack.pop()
elif op == 'JUMP':
pc = stack.pop()
elif op == 'LEAVE' and halt_on_leave:
break # bounded helper return boundary; no real VM frame is unwound
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()
|