| |
| """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() |
|
|
