XFEEDToolBox

"""Close current trade-flow readers across fixed images without executing the game."""
import hashlib
import importlib.util
import json
import re
import struct
from pathlib import Path

spec = importlib.util.spec_from_file_location('audit', Path(__file__).with_name('Audit-GalaxyClientCompatibility.py'))
audit = importlib.util.module_from_spec(spec)
spec.loader.exec_module(audit)
old = audit.Image(Path('artifacts/ship-model-reference/client-copy/EliteDangerous64.exe'))
new = audit.Image(Path('D:/SteamLibrary/steamapps/common/Elite Dangerous/Products/elite-dangerous-odyssey-64/EliteDangerous64.exe'))
functions, references = [], []
for before in (0x3AD9300, 0x3AC9700, 0x3B4E2A0, 0x3AF8980, 0x3B92120,
               0x3B55BB0, 0x3B2ABE0, 0x3AB9960, 0x3B92B40, 0x99F1A0, 0x9A0F30):
    span = old.function(before)
    if not span or span[0] != before:
        raise ValueError(f'Missing bounded original function {before:x}')
    row = audit.match(old, new, before, span[1] - before)
    if row['status'] != 'unique':
        raise ValueError(f'Unclosed trade-flow function {before:x}: {row["status"]}')
    row['oldRva'] = hex(before)
    functions.append(row)
    references.extend(audit.references(old, new, before, int(row['newRva'], 16), row['length']))

tables = {}
for before in (0x552EAA0, 0x552EAD0, 0x552EBD0, 0x552CE58, 0x552CE88, 0x552CF88):
    candidates = {row[1] for row in references if row[0] == before}
    if len(candidates) != 1:
        raise ValueError(f'Trade-flow table binding ambiguous: {before:x}')
    tables[before] = candidates.pop()

globals_ = {}
for before in (0x5F05CA8, 0x5F856E0):
    candidates = {row[1] for row in references if row[0] == before}
    if len(candidates) != 1:
        raise ValueError(f'Trade-flow source ID binding ambiguous: {before:x}')
    globals_[before] = candidates.pop()

# The short type getter is not uniquely signature-matchable after masking RIP.
# Bind it instead by its actual paired dynamic clock type ID, then its unique
# readonly vtable pointer. This does not assume a global table shift.
clock_getters = []
for section in new.pe.sections:
    if not section.Characteristics & 0x20000000:
        continue
    for match in re.finditer(rb'\x8b\x05....\x89\x02\x48\x8b\xc2\xc3', section.get_data(), re.S):
        address = section.VirtualAddress + match.start()
        target = address + 6 + struct.unpack_from('<i', match.group(), 2)[0]
        if target == globals_[0x5F05CA8]:
            clock_getters.append(address)
if len(clock_getters) != 1:
    raise ValueError('Clock type getter ambiguous')
clock_tables = []
for section in new.pe.sections:
    if section.Characteristics & 0x20000000:
        continue
    data, pointer, offset = section.get_data(), struct.pack('<Q', new.base + clock_getters[0]), 0
    while (offset := data.find(pointer, offset)) >= 0:
        clock_tables.append(section.VirtualAddress + offset)
        offset += 8
if len(clock_tables) != 1:
    raise ValueError('Clock interface table ambiguous')
tables[0x5134AE0] = clock_tables[0]
clock_ctor_tables = {row[1] for row in references if row[0] == 0x5134AE0}
if clock_ctor_tables != {clock_tables[0]}:
    raise ValueError('Clock getter/table does not match paired constructor and destructor operands')

slots = []
for table, offset in ((0x552EBD0, 0), (0x552EBD0, 0x60), (0x552EBD0, 0x68),
                      (0x552EBD0, 0x70), (0x552CF88, 0), (0x5134AE0, 0), (0x5134AE0, 0x28)):
    before = struct.unpack('<Q', old.data(table + offset, 8))[0] - old.base
    after = struct.unpack('<Q', new.data(tables[table] + offset, 8))[0] - new.base
    length = next(i.address + i.size - old.base - before for i in old.instructions(before, 512) if i.mnemonic == 'ret')
    raw, mask, *_ = audit.signature(old, before, length)
    actual = new.data(after, length)
    if any(a != b for a, b, keep in zip(raw, actual, mask) if keep):
        raise ValueError(f'Trade-flow interface slot changed {table:x}+{offset:x}')
    paired = audit.references(old, new, before, after, length)
    references.extend(paired)
    slots.append(dict(oldVtable=hex(table), newVtable=hex(tables[table]), slot=hex(offset),
                      oldRva=hex(before), newRva=hex(after), bytes=length,
                      oldSha256=hashlib.sha256(raw).hexdigest(), newSha256=hashlib.sha256(actual).hexdigest(),
                      references=[list(map(hex, row)) for row in paired]))

map_ids = {row[1] for row in references if row[0] == 0x5F87B20}
if len(map_ids) != 1:
    raise ValueError('Trade map type ID ambiguous')
globals_[0x5F87B20] = map_ids.pop()
result = dict(runtimeNeutral=True, canInstallHooks=False, gameCalls=0, networkRequests=0,
              oldImage=dict(path=str(old.path), sha256=hashlib.sha256(old.raw).hexdigest()),
              newImage=dict(path=str(new.path), sha256=hashlib.sha256(new.raw).hexdigest()),
              functions=functions, vtables={hex(a): hex(b) for a, b in tables.items()}, slots=slots,
              globals={hex(a): hex(b) for a, b in globals_.items()})
path = Path('artifacts/galaxy-query-research/account-sources/current-trade-compatibility.json')
path.write_text(json.dumps(result, indent=2) + '\n', encoding='utf-8')
print(json.dumps(dict(functions=len(functions), slots=len(slots), vtables=result['vtables'], globals=result['globals']), indent=2))