File size: 7,034 Bytes
0c343f8 | 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 | #!/usr/bin/env python3
"""Recover native event queue/address evidence; inspect bounded ELF64 journals."""
import argparse
import gzip
import hashlib
import json
from pathlib import Path
import re
import struct
from analyze_engine_control import control_flow, windows
from analyze_frame_timing import read_address
from analyze_native import elf
from analyze_qvm import write_json
from analyze_vm_extensions import native_functions, verified
EVENT = struct.Struct('<iiiii4sQ')
def parse_journal(data, max_events=100000, max_payload=16*1024*1024):
"""Inspect this observed little-endian ELF64 layout, never dereference pointers.
Bounds are inspector policy, not recovered original-engine validation rules.
The historical pointer and padding are preserved as evidence, not interpreted
as a usable address or a portable file-format guarantee.
"""
if max_events < 1 or max_payload < 0:
raise ValueError('invalid journal inspection limits')
records, offset = [], 0
while offset < len(data):
if len(records) >= max_events or len(data)-offset < EVENT.size:
raise ValueError('journal record limit or truncated header')
time, kind, value, value2, size, padding, pointer = EVENT.unpack_from(data, offset)
if size < 0 or size > max_payload or size > len(data)-offset-EVENT.size:
raise ValueError('negative, oversized or truncated payload')
start = offset + EVENT.size
payload = data[start:start+size]
records.append({'index': len(records), 'file_offset': offset, 'time_signed32': time,
'type_signed32': kind, 'value_signed32': value, 'value2_signed32': value2,
'payload_bytes': size, 'padding_hex': padding.hex(), 'historical_pointer': pointer,
'payload_sha256': hashlib.sha256(payload).hexdigest()})
offset = start+size
return {'layout': 'observed Linux ELF64 little-endian 32-byte event header plus payload',
'bytes': len(data), 'sha256': hashlib.sha256(data).hexdigest(), 'records': records,
'limitations': 'Inspector policy, not the original engine parser. Types and pointer values '
'are retained without validation; zero-length payloads consume no bytes.'}
def immediate_strings(data, sections, graph):
result = []
for ins in graph['instructions']:
if ins['opcode'] != 'movl':
continue
match = re.match(r'\$0x([0-9a-f]+), %edx(?:\s|$)', ins['operands'])
if not match:
continue
address = int(match[1], 16)
raw = bytearray()
for at in range(address, address+256):
chunk, _ = read_address(data, sections, at, 1)
if chunk == b'\0':
break
raw.extend(chunk)
else:
raise ValueError('unterminated address-format string')
result.append({'instruction': ins['address'], 'address': address, 'text': raw.decode('ascii')})
return result
def symbol_record(symbols, name):
matches = {(s['address'], s['bytes']) for s in symbols if s['name'] == name and not s['undefined']}
if len(matches) != 1:
raise ValueError('missing or ambiguous queue symbol: ' + name)
address, size = matches.pop()
return {'name': name, 'address': address, 'bytes': size}
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--journal', type=Path, help='inspect a captured Linux ELF64 journal instead of native binaries')
parser.add_argument('--output', type=Path, help='JSON destination for journal inspection')
parser.add_argument('--max-payload', type=int, default=16*1024*1024)
args = parser.parse_args()
if args.journal:
if not args.output:
parser.error('--journal requires --output')
if args.output.resolve() == args.journal.resolve():
parser.error('journal output must not overwrite its input')
if args.journal.stat().st_size > 512*1024*1024:
raise ValueError('journal exceeds inspector file-size limit')
write_json(args.output, parse_journal(args.journal.read_bytes(), max_payload=args.max_payload))
return
if args.output:
parser.error('--output requires --journal')
output = {'native': {}, 'limitations': 'Static Linux ELF64 only. No original game process or native '
'callee is executed. Literal offset references are not complete alias analysis; '
'queue counter overflow and runtime journal compatibility remain unverified.'}
names = ['Com_PushEvent', 'Com_Milliseconds', 'Com_InitPushEvent', 'Sys_QueEvent',
'NET_AdrToString', 'NET_IsLocalAddress', 'Netchan_Setup', 'SV_BotAllocateClient']
queue_names = ['eventQue', 'eventHead', 'eventTail', 'com_pushedEvents',
'com_pushedEventsHead', 'com_pushedEventsTail']
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, symbols = elf(data)
graphs = {name: control_flow(functions[name]) for name in names}
references = [{'function': name, 'function_address': fn['address'], 'line': line.strip()}
for name, fn in functions.items() for line in fn['lines']
if re.search(r'0x24ee4\(', line)]
focused = ['SV_DirectConnect', 'SV_UserinfoChanged', 'SV_PacketEvent', 'SV_DropClient']
output['native'][label] = {'source': report['source'], 'disassembly_sha256': artifact['uncompressed_sha256'],
'graphs': graphs, 'queue_symbols': [symbol_record(symbols, n) for n in queue_names],
'address_format_literals': immediate_strings(data, metadata['sections'], graphs['NET_AdrToString']),
'client_address_type_literal_references': references,
'client_address_windows': {name: windows(functions[name], [r'0x24ee4\(', r'<Netchan_Setup>', r'<SV_BotFreeClient>'], 15, 9) for name in focused},
'journal_header_layout': {'bytes': EVENT.size, 'endianness': 'little', 'fields': {
'time': [0, 4], 'type': [4, 4], 'value': [8, 4], 'value2': [12, 4],
'payload_length': [16, 4], 'padding': [20, 4], 'historical_pointer': [24, 8]}}}
write_json(Path('docs/research/event-queues.json'), output)
print('Recovered sixteen native queue/address graphs and bounded ELF64 journal inspector')
if __name__ == '__main__':
main()
|