XFEEDToolBox

"""Create/verify runtime-neutral offline mapping evidence; never installs hooks."""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
from pathlib import Path
import re


def audit_module():
    path = Path(__file__).with_name('Audit-GalaxyClientCompatibility.py')
    spec = importlib.util.spec_from_file_location('galaxy_pe_audit', path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def internal_branches(module, old, new, before, after, length):
    a, b = old.instructions(before, length), new.instructions(after, length)
    if len(a) != len(b):
        raise ValueError('Instruction boundaries changed')
    count = 0
    for left, right in zip(a, b):
        if left.size != right.size or left.mnemonic != right.mnemonic:
            raise ValueError('Instruction layout changed')
        if not (left.group(module.cs.CS_GRP_JUMP) or left.group(module.cs.CS_GRP_CALL)):
            continue
        for x, y in zip(left.operands, right.operands):
            if x.type == y.type == module.X86_OP_IMM and old.base + before <= x.imm < old.base + before + length:
                if x.imm - old.base - before != y.imm - new.base - after:
                    raise ValueError('Internal control-flow target changed')
                count += 1
    return count


def code_record(module, old, new, before, after, method, sources=None, limit=384):
    signature = module.signature(old, before, limit)
    if not signature:
        raise ValueError(f'No signature for {before:x}')
    raw, mask, *_ = signature
    actual = new.data(after, len(raw))
    if len(actual) != len(raw) or any(a != b for a, b, keep in zip(raw, actual, mask) if keep):
        raise ValueError(f'Code signature mismatch {before:x}->{after:x}')
    branches = internal_branches(module, old, new, before, after, len(raw))
    return {'oldRva': hex(before), 'newRva': hex(after), 'kind': 'code', 'method': method,
            'sources': sources or [], 'bytes': len(raw),
            'internalBranchesVerified': branches,
            'maskHex': mask.hex(), 'oldBytesSha256': hashlib.sha256(raw).hexdigest(),
            'newBytesSha256': hashlib.sha256(actual).hexdigest()}


def data_record(old, new, before, after, method, sources=None):
    section = old.pe.get_section_by_rva(before)
    target = new.pe.get_section_by_rva(after)
    if not section or not target or section.Name != target.Name:
        raise ValueError(f'Data section mismatch {before:x}->{after:x}')
    initialised = before - section.VirtualAddress < section.SizeOfRawData
    new_initialised = after - target.VirtualAddress < target.SizeOfRawData
    length = 16 if initialised and new_initialised else 0
    return {'oldRva': hex(before), 'newRva': hex(after), 'kind': 'data', 'method': method,
            'sources': sources or [], 'section': section.Name.rstrip(b'\0').decode(),
            'storage': 'file-backed' if length else 'zero-fill-runtime-value-unknown',
            'bytes': length, 'oldBytesSha256': hashlib.sha256(old.data(before, length)).hexdigest(),
            'newBytesSha256': hashlib.sha256(new.data(after, length)).hexdigest()}


def manual_code_record(module, old, new, row):
    before, after, length = int(row['oldRva'], 16), int(row['newRva'], 16), row['bytes']
    raw, actual = old.data(before, length), new.data(after, length)
    sig = module.signature(old, before, length)
    mask = bytearray(sig[1]) + bytearray(b'\xff' * (length - len(sig[1])))
    for field in row['extraAddressFields']:
        at, width = field['offset'], field.get('width', 4)
        if int.from_bytes(raw[at:at + width], 'little') != int(field['oldValue'], 16) or \
           int.from_bytes(actual[at:at + width], 'little') != int(field['newValue'], 16):
            raise ValueError('Manual address field changed')
        mask[at:at + width] = b'\0' * width
    if len(raw) != length or len(actual) != length or any(a != b for a, b, keep in zip(raw, actual, mask) if keep):
        raise ValueError('Manual body has unexplained differences')
    return {'oldRva': row['oldRva'], 'newRva': row['newRva'], 'kind': 'code',
            'method': row.get('method', 'manually-reviewed-embedded-address-fields'), 'evidence': row['basis'],
            'sources': row.get('sources', []), 'bytes': length, 'maskHex': mask.hex(),
            'oldBytesSha256': hashlib.sha256(raw).hexdigest(), 'newBytesSha256': hashlib.sha256(actual).hexdigest()}


def published_galaxy_bindings(module, old, new):
    """Recheck later read-only adapters against their original paired bodies.

    These small sidecars extend the original source inventory. In particular,
    zero-filled globals are established by instruction operands, not by equal
    zero bytes or a presumed image-wide displacement.
    """
    root = Path('artifacts/galaxy-query-research')
    paths = [root / 'power-outer-sources/compatibility.json',
             root / 'thargoid-account-sources/compatibility.json',
             root / 'power-outer-sources/territory-constants.json',
             root / 'fleet-mission-sources/compatibility.json',
             root / 'fleet-mission-sources/missions-compatibility.json',
             root / 'commodity-tree/compatibility.json',
             root / 'account-sources/current-trade-compatibility.json',
             root / 'constellations/quality-provider-compatibility.json',
             Path('artifacts/galaxy-client-compatibility/local-docking-slot-closure.json'),
             root / 'constellations/catalogue-profile-compatibility.json',
             root / 'allegiance-definitions/compatibility.json',
             Path('artifacts/galaxy-client-compatibility/community-goal-optin-compatibility.json'),
             root / 'navroute-membership/compatibility.json',
             root / 'route-settings/compatibility.json',
             root / 'point-overrides/compatibility.json',
             root / 'route-filter/compatibility.json',
             root / 'civilization/compatibility.json',
             root / 'service-facts/compatibility.json',
             root / 'carrier-facts/compatibility.json',
             root / 'power-outer-sources/background/dust-runtime/compatibility.json',
             root / 'route-checkbox-filters/compatibility.json',
             root / 'power-outer-extent/compatibility.json',
             root / 'state-facts/compatibility.json',
             Path('artifacts/galaxy-client-compatibility/docking-zero-update-compatibility.json'),
             root / 'colonisation-colour/compatibility.json',
             root / 'colonisation-highlight/compatibility.json',
             root / 'power-influence/compatibility.json']
    proofs = [json.loads(path.read_text(encoding='utf8')) for path in paths]
    old_hash, new_hash = (hashlib.sha256(image.raw).hexdigest() for image in (old, new))
    records, references = {}, []
    for proof, path in zip(proofs, paths):
        route_schema=path==paths[20]
        source_hash_schema=route_schema or path in (paths[24],paths[25],paths[26])
        if source_hash_schema:
            if proof.get('gameProcessAccess') is not False or proof.get('runtimeValidated') is not False or \
               proof.get('sourceHashes')!=[old_hash,new_hash]:
                raise ValueError('Original route-mask proof scope/images changed')
        elif proof.get('runtimeNeutral') is not True:
            raise ValueError('Published binding proof is not runtime-neutral')
        for prefix, expected in (('old', old_hash), ('new', new_hash)):
            actual = proof['sourceHashes'][0 if prefix=='old' else 1] if source_hash_schema else proof.get(prefix + 'ImageSha256', proof.get(prefix + 'Image', {}).get('sha256'))
            if actual != expected:
                raise ValueError('Published binding proof image mismatch: ' + path.name)
        for row in proof['functions']:
            if row['status'] != 'unique':
                raise ValueError('Published binding function remains ambiguous')
            before = int(row.get('oldRva', row.get('old', row.get('oldFunctionRange', [''])[0])), 16)
            after = int(row.get('newRva', row.get('new', '')), 16)
            length = row.get('length', row.get('bytes'))
            original_function=old.function
            if route_schema and row.get('boundedLeafReason'):
                # Its absolute indexed jump table is inside this explicitly
                # bounded switch leaf. The general no-unwind96B fallback is
                # insufficient; validate through the unique physical RET.
                ins=old.instructions(before,length)
                returns=[i for i in ins if i.group(module.cs.CS_GRP_RET)]
                if original_function(before) is not None or len(returns)!=1 or returns[0].address+returns[0].size-old.base!=before+length:
                    raise ValueError('Original route switch leaf bound changed')
                old.function=lambda rva,fn=original_function,start=before,end=before+length:(start,end,0) if start<=rva<end else fn(rva)
            try:
                records[hex(before)] = code_record(module, old, new, before, after,
                    'published-adapter-full-body-and-operand-binding', [str(path)], limit=length)
            finally:old.function=original_function
            for table in row.get('actualSwitchTables',[]):
                expected=table['relativeTargets']
                if len(expected)!=16:raise ValueError('Published original classification switch size changed')
                for image,prefix,start in ((old,'old',before),(new,'new',after)):
                    target=int(table[prefix+'Table'],16)
                    actual=[int.from_bytes(image.data(target+index*4,4),'little')-start for index in range(16)]
                    if actual!=expected or any(offset<0 or offset>=length for offset in actual):
                        raise ValueError('Published original classification switch targets changed')
            if row.get('rawLeafWholeBody'):
                # A leaf may return early without PE unwind metadata. Validate
                # its complete bounded body, including all later hash buckets,
                # rather than recording only the first return's signature.
                raw, actual = old.data(before,length),new.data(after,length)
                if raw!=actual or hashlib.sha256(raw).hexdigest()!=row['oldWindowSha256'] or hashlib.sha256(actual).hexdigest()!=row['newWindowSha256']:
                    raise ValueError('Published full multi-return leaf changed')
                records[hex(before)].update(bytes=length,maskHex='ff'*length,
                    oldBytesSha256=hashlib.sha256(raw).hexdigest(),newBytesSha256=hashlib.sha256(actual).hexdigest(),
                    method='published-typed-slot-complete-identical-multi-return-leaf',
                    internalBranchesVerified=internal_branches(module,old,new,before,after,length))
            references.extend(module.references(old, new, before, after, length))

    outer, thargoid, territory, fleet, missions, commodities, tradeflows, quality, docking, catalogue_profile, allegiance, community_goal, navroute, route_settings, point_overrides, route_filter, civilization, service_facts, carrier_facts, dust_runtime, route_masks, outer_extent, state_facts, docking_zero_update, colonisation_colour, colonisation_highlight, power_influence = proofs
    # The route-mask schema calls its actual typed-slot rows `vtables` and
    # uses oldTarget/newTarget. Preserve and validate those actual slots;
    # they are not a list of presumed base-vtable addresses.
    route_slots=[dict(oldVtable=row['oldVtable'],newVtable=row['newVtable'],slot=row['slot'],oldRva=row['oldTarget'],newRva=row['newTarget']) for row in route_masks['vtables']]
    route_typed=dict(slots=route_slots,vtables={row['oldVtable']:row['newVtable'] for row in route_masks['providers']})
    dispatch=next(row for row in route_masks['functions'] if int(row['oldRva'],16)==0x3B16B00)
    dispatch_targets=[]
    for image,start in ((old,0x3B16B00),(new,int(dispatch['newRva'],16))):
        candidates=[operand for ins in image.instructions(start,64) for operand in ins.operands if operand.type==module.X86_OP_MEM and operand.mem.index!=0]
        if len(candidates)!=1:raise ValueError('Original route category dispatch table is ambiguous')
        targets=[int.from_bytes(image.data(candidates[0].mem.disp+index*4,4),'little') for index in range(16)]
        if any(target<start or target>=start+dispatch['length'] for target in targets):raise ValueError('Original route category target leaves its verified leaf')
        dispatch_targets.append([target-start for target in targets])
    if dispatch_targets[0]!=dispatch_targets[1]:raise ValueError('Original route category jump targets changed')
    for row in outer['navigationGetters']:
        before, after, length = int(row['oldRva'], 16), int(row['newRva'], 16), row['bytes']
        for image, prefix, expected in ((old, 'old', before), (new, 'new', after)):
            actual = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16) + int(row['slot'], 16), 8), 'little') - image.base
            if actual != expected:
                raise ValueError('Published navigation getter slot changed')
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-original-navigation-vtable-getter', [str(paths[0])], limit=length)

    def bound_data(before, after, basis):
        pairs = [r for r in references if r[0] == before]
        if not pairs or {r[1] for r in pairs} != {after}:
            raise ValueError('Published data binding is not established by paired operands: ' + hex(before))
        record = data_record(old, new, before, after, 'published-ctor-or-getter-paired-operand', basis)
        record['evidence'] = [list(map(hex, row)) for row in pairs]
        records[hex(before)] = record

    for row in quality['getters']:
        before, after, offset = int(row['oldRva'],16), int(row['newRva'],16), int(row['slot'],16)
        for image, prefix, expected in ((old,'old',before),(new,'new',after)):
            actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
            if actual!=expected:raise ValueError('Published quality-provider getter slot changed')
        records[hex(before)]=code_record(module,old,new,before,after,
            'published-actual-quality-provider-getter',[str(paths[7])],limit=row['bytes'])
        references.extend(module.references(old,new,before,after,row['bytes']))
    for before, after in quality['vtables'].items():
        bound_data(int(before,16),int(after,16),[str(paths[7])])
    for before, after in quality['globals'].items():
        bound_data(int(before,16),int(after,16),[str(paths[7])])

    # The availability leaf newly referenced by LocalFlight is identified by
    # its actual typed docking interface, never a generic adjacent getter.
    for row in docking['slots']:
        before, after, offset = int(row['oldRva'],16), int(row['newRva'],16), int(row['slot'],16)
        for image, prefix, expected in ((old,'old',before),(new,'new',after)):
            actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
            if actual!=expected:raise ValueError('Published typed docking getter slot changed')
        records[hex(before)]=code_record(module,old,new,before,after,
            'published-original-typed-docking-interface-slot',[str(paths[8])],limit=row['bytes'])
        references.extend(module.references(old,new,before,after,row['bytes']))
    for group in ('vtables','globals'):
        for before, after in docking[group].items():
            bound_data(int(before,16),int(after,16),[str(paths[8])])

    for row in catalogue_profile['slots']:
        before, after, offset = int(row['oldRva'],16), int(row['newRva'],16), int(row['slot'],16)
        for image, prefix, expected in ((old,'old',before),(new,'new',after)):
            actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
            if actual!=expected:raise ValueError('Published catalogue profile getter slot changed')
        records[hex(before)]=code_record(module,old,new,before,after,
            'published-catalogue-renderer-original-field-getter',[str(paths[9])],limit=row['bytes'])
        references.extend(module.references(old,new,before,after,row['bytes']))
    for group in ('vtables','globals'):
        for before, after in catalogue_profile[group].items():
            bound_data(int(before,16),int(after,16),[str(paths[9])])

    for row in allegiance['slots']:
        before, after, offset = int(row['oldRva'],16), int(row['newRva'],16), int(row['slot'],16)
        for image, prefix, expected in ((old,'old',before),(new,'new',after)):
            actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
            if actual!=expected:raise ValueError('Published allegiance definition getter slot changed')
        if records.get(hex(before),{}).get('method')!='published-typed-slot-complete-identical-multi-return-leaf':
            records[hex(before)]=code_record(module,old,new,before,after,
                'published-allegiance-definition-typed-interface-slot',[str(paths[10])],limit=row['bytes'])
        references.extend(module.references(old,new,before,after,row['bytes']))
    for row in navroute['slots']:
        before, after, offset = int(row['oldRva'],16), int(row['newRva'],16), int(row['slot'],16)
        for image, prefix, expected in ((old,'old',before),(new,'new',after)):
            actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
            if actual!=expected:raise ValueError('Published NavRoute getter typed slot changed')
        records[hex(before)]=code_record(module,old,new,before,after,
            'published-navroute-actual-typed-slot',[str(paths[12])],limit=row['bytes'])
        references.extend(module.references(old,new,before,after,row['bytes']))
    binding=navroute['callBinding']
    for image,prefix in ((old,'old'),(new,'new')):
        actual=image.instructions(int(binding[prefix+'Call'],16),5)
        if len(actual)!=1 or actual[0].mnemonic!='call' or actual[0].operands[0].type!=module.X86_OP_IMM or actual[0].operands[0].imm-image.base!=int(binding[prefix+'Callee'],16):
            raise ValueError('Published NavRoute membership actual callee changed')
    for group in ('vtables','globals'):
        for before, after in allegiance[group].items():
            bound_data(int(before,16),int(after,16),[str(paths[10])])

    # These adapters add actual typed slots, including the small route type-ID
    # getter. Importing only their larger functions would leave those guarded
    # source addresses uninventoried and would not verify interface identity.
    for proof, path in ((route_settings,paths[13]),(point_overrides,paths[14]),(route_filter,paths[15]),
                        (civilization,paths[16]),(service_facts,paths[17]),(carrier_facts,paths[18]),
                        (dust_runtime,paths[19]),(route_typed,paths[20]),(outer_extent,paths[21]),(state_facts,paths[22]),(colonisation_colour,paths[24]),(colonisation_highlight,paths[25]),(power_influence,paths[26])):
        for row in proof.get('slots',[]):
            before,after,offset=int(row['oldRva'],16),int(row['newRva'],16),int(row['slot'],16)
            for image,prefix,expected in ((old,'old',before),(new,'new',after)):
                actual=int.from_bytes(image.data(int(row[prefix+'Vtable'],16)+offset,8),'little')-image.base
                if actual!=expected:raise ValueError('Published adapter typed slot changed: '+str(path))
            span=old.function(before)
            length=row.get('bytes',row.get('length')) or records.get(hex(before),{}).get('bytes')
            if length is None:
                if span is None or span[0]!=before:raise ValueError('Published route/point slot needs an exact body bound')
                length=span[1]-before
            records[hex(before)]=code_record(module,old,new,before,after,
                'published-adapter-actual-typed-slot',[str(path)],limit=length)
            if row.get('rawLeafWholeBody'):
                raw,actual=old.data(before,length),new.data(after,length)
                if raw!=actual:raise ValueError('Published complete typed leaf changed')
                records[hex(before)].update(bytes=length,maskHex='ff'*length,
                    oldBytesSha256=hashlib.sha256(raw).hexdigest(),newBytesSha256=hashlib.sha256(actual).hexdigest(),
                    method='published-actual-typed-slot-complete-identical-leaf',
                    internalBranchesVerified=internal_branches(module,old,new,before,after,length))
            references.extend(module.references(old,new,before,after,length))
        for group in ('vtables','globals'):
            bindings=proof.get(group,{})
            pairs=bindings.items() if isinstance(bindings,dict) else ((row['oldRva'],row['newRva']) for row in bindings)
            for before,after in pairs:
                bound_data(int(before,16),int(after,16),[str(path)])
    service_type=service_facts['typeGlobal']
    bound_data(int(service_type['oldRva'],16),int(service_type['newRva'],16),[str(paths[17])])

    for row in community_goal['callBindings']:
        before,after=int(row['oldFunction'],16),int(row['newFunction'],16)
        if records[hex(before)]['newRva']!=hex(after):raise ValueError('CG function binding changed')
        callee=records.get(row['oldCallee'])
        if not callee or callee['newRva']!=row['newCallee']:raise ValueError('CG independent send helper changed')
        if int(row['oldCall'],16)-before!=int(row['newCall'],16)-after:raise ValueError('CG call position changed')
        for image,prefix,function in ((old,'old',before),(new,'new',after)):
            at=int(row[prefix+'Call'],16);target=int(row[prefix+'Callee'],16)
            if not function<=at<function+records[hex(before)]['bytes']:raise ValueError('CG call is outside the paired body')
            ins=image.instructions(at,5)
            if len(ins)!=1 or ins[0].mnemonic!='call' or len(ins[0].operands)!=1 or ins[0].operands[0].type!=module.X86_OP_IMM or ins[0].operands[0].imm-image.base!=target:
                raise ValueError('CG independently proved direct call changed')
    request_type=community_goal['requestType']
    for image,prefix in ((old,'old'),(new,'new')):
        endpoint=(request_type['endpoint']+'\0').encode('utf8')
        if image.data(int(request_type[prefix+'Endpoint'],16),len(endpoint))!=endpoint:raise ValueError('CG opt-in endpoint changed')
        expected=b'\xba'+int(request_type['kind']).to_bytes(4,'little')
        if image.data(int(request_type[prefix+'KindInstruction'],16),5)!=expected:raise ValueError('CG constructor request kind changed')
    for field in ('Vtable','Endpoint'):
        bound_data(int(request_type['old'+field],16),int(request_type['new'+field],16),[str(paths[11])])

    for before in (0x55276E8, 0x5527718, 0x5527840, 0x55278C0):
        bound_data(before, int(thargoid['vtables'][hex(before)], 16), [str(paths[1])])
    for group, old_table in (('publishedInterfaceSlots', 0x55278C0),
                             ('sharedCommunityGoalInterfaceSlots', 0x5527840)):
        new_table = int(thargoid['vtables'][hex(old_table)], 16)
        for row in thargoid[group]:
            before, after, offset = int(row['oldRva'], 16), int(row['newRva'], 16), int(row['slot'], 16)
            for image, table, expected in ((old, old_table, before), (new, new_table, after)):
                actual = int.from_bytes(image.data(table + offset, 8), 'little') - image.base
                if actual != expected:
                    raise ValueError('Published account getter slot changed')
            records[hex(before)] = code_record(module, old, new, before, after,
                'published-account-interface-getter', [str(paths[1])], limit=row['bytes'])
            references.extend(module.references(old, new, before, after, row['bytes']))
    for address in (0x5F5428C, 0x5FA10F0):
        bound_data(address, address, [str(paths[1])])
    for key in ('overrideColourAlpha', 'overrideGeometry'):
        row = territory['fields'][key]
        bound_data(int(row['oldRva'], 16), int(row['newRva'], 16), [str(paths[2])])
    for row in fleet['slots']:
        before, after, offset = int(row['oldRva'], 16), int(row['newRva'], 16), int(row['slot'], 16)
        for image, prefix, expected in ((old, 'old', before), (new, 'new', after)):
            actual = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16) + offset, 8), 'little') - image.base
            if actual != expected:
                raise ValueError('Published Fleet getter slot changed')
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-fleet-interface-getter', [str(paths[3])], limit=row['bytes'])
        references.extend(module.references(old, new, before, after, row['bytes']))
    for before, after in fleet['vtables'].items():
        if before == '0x5294d08':
            # This table is independently closed by both original map-used
            # actor methods, rather than by a constructor in this sidecar.
            for offset, function in ((0x80, 0x2561450), (0xE0, 0x255A140)):
                paired = records[hex(function)]
                for image, table, method in ((old, int(before, 16), function),
                    (new, int(after, 16), int(paired['newRva'], 16))):
                    if int.from_bytes(image.data(table + offset, 8), 'little') - image.base != method:
                        raise ValueError('Published Fleet actor method binding changed')
            record = data_record(old, new, int(before, 16), int(after, 16),
                'published-map-actor-two-method-vtable-binding', [str(paths[3])])
            record['evidence'] = ['vfunc80:2561450', 'vfuncE0:255A140']
            records[before] = record
        else:
            bound_data(int(before, 16), int(after, 16), [str(paths[3])])
    for before, after in fleet['globals'].items():
        bound_data(int(before, 16), int(after, 16), [str(paths[3])])
    for row in missions['slots']:
        before, after, offset = int(row['oldRva'], 16), int(row['newRva'], 16), int(row['slot'], 16)
        for image, prefix, expected in ((old, 'old', before), (new, 'new', after)):
            actual = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16) + offset, 8), 'little') - image.base
            if actual != expected:
                raise ValueError('Published mission getter slot changed')
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-mission-interface-getter', [str(paths[4])], limit=row['bytes'])
        references.extend(module.references(old, new, before, after, row['bytes']))
    for before, after in missions['vtables'].items():
        bound_data(int(before, 16), int(after, 16), [str(paths[4])])
    for before, after in missions['globals'].items():
        bound_data(int(before, 16), int(after, 16), [str(paths[4])])
    for row in commodities['slots']:
        before, after, offset = int(row['oldRva'], 16), int(row['newRva'], 16), int(row['slot'], 16)
        for image, prefix, expected in ((old, 'old', before), (new, 'new', after)):
            actual = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16) + offset, 8), 'little') - image.base
            if actual != expected:
                raise ValueError('Published commodity getter slot changed')
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-commodity-interface-getter', [str(paths[5])], limit=row['bytes'])
        references.extend(module.references(old, new, before, after, row['bytes']))
    for row in commodities['getters']:
        before, after = int(row['oldRva'], 16), int(row['newRva'], 16)
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-commodity-name-and-category-getter', [str(paths[5])], limit=row['bytes'])
        references.extend(module.references(old, new, before, after, row['bytes']))
    for group in ('vtables', 'globals'):
        for before, after in commodities[group].items():
            bound_data(int(before, 16), int(after, 16), [str(paths[5])])
    for row in tradeflows['slots']:
        before, after, offset = int(row['oldRva'], 16), int(row['newRva'], 16), int(row['slot'], 16)
        for image, prefix, expected in ((old, 'old', before), (new, 'new', after)):
            actual = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16) + offset, 8), 'little') - image.base
            if actual != expected:
                raise ValueError('Published current trade getter slot changed')
        records[hex(before)] = code_record(module, old, new, before, after,
            'published-current-trade-interface-getter', [str(paths[6])], limit=row['bytes'])
        references.extend(module.references(old, new, before, after, row['bytes']))
    for group in ('vtables', 'globals'):
        for before, after in tradeflows[group].items():
            bound_data(int(before, 16), int(after, 16), [str(paths[6])])
    artifacts = [{'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} for path in paths]
    ring_path,ring_dependencies=verify_original_ring_geometry(old)
    artifacts.extend({'path':str(path),'sha256':hashlib.sha256(path.read_bytes()).hexdigest()} for path in [ring_path,*ring_dependencies])
    return records, artifacts


def verify_original_ring_geometry(old):
    """Pinned geometry/shader proof is not an ED5 Native address mapping."""
    root=Path('artifacts/galaxy-query-research/power-outer-sources')
    path=root/'focus-ring/original-ring.json'
    proof=json.loads(path.read_text(encoding='utf8'))
    if proof.get('gameProcessAccess') is not False or proof.get('gameCalls')!=0 or proof.get('networkRequests')!=0 or \
       proof['imageSha256']!=hashlib.sha256(old.raw).hexdigest():raise ValueError('Original range geometry scope/image changed')
    for row in proof['functions']:
        if hashlib.sha256(old.data(int(row['rva'],16),row['length'])).hexdigest()!=row['sha256']:
            raise ValueError('Original range geometry body changed')
    if proof.get('primitive')!='TriangleStrip' or proof.get('primitiveCount')!=32 or proof.get('vertexCount')!=34 or \
       old.data(0x4E2DE48+5*2,2)!=b'\x01\x02' or int.from_bytes(old.data(0x4E2DE78+5*4,4),'little')!=5:
        raise ValueError('Original range topology/count proof changed')
    import struct
    def floats(at,count):return list(struct.unpack('<'+'f'*count,old.data(at,count*4)))
    constants=dict(distanceMaximum=floats(0x4E29750,1)[0],distanceMinimum=floats(0x4E328C0,1)[0],widthScale=floats(0x552F4E8,1)[0],
                   uAxis=floats(0x50CB0D0,3),vAxis=floats(0x50CB0F0,3),colour=floats(0x54C8330,4))
    if constants!=proof['constants']:raise ValueError('Original range geometry constants changed')
    material_path=root/'material/original-materials.json';blend_path=root/'material-slots/original-blend-application.json'
    material=next(row for row in json.loads(material_path.read_text())['materials'] if row['name']=='focusring')
    state=next(row for row in json.loads(blend_path.read_text())['materials'] if row['material']=='focusring')
    if material!=proof['material'] or not state['resolved'] or state['originalEnums']!=proof['sourceBlend']:
        raise ValueError('Original range material/state changed')
    dependencies=[material_path,blend_path]
    for stage,expected in proof['shaderSha256'].items():
        shader=root/f'effects/{stage}_GalaxyMapFocusRing_Forward_0_Win64_SM50.dxbc';dependencies.append(shader)
        if hashlib.sha256(shader.read_bytes()).hexdigest()!=expected:raise ValueError('Original range shader changed')
    if proof['exposure'].get('runtimeResolved') is not False or proof['exposure'].get('unknownMustRemainHidden') is not True:
        raise ValueError('Original range unresolved exposure must remain explicit')
    return path,dependencies


def build(audit_path, output):
    module = audit_module()
    report = json.loads(audit_path.read_text(encoding='utf8'))
    old = module.Image(Path(report['old']['path']))
    new = module.Image(Path(report['new']['path']))
    mappings, unresolved = {}, []
    for key, value in report['functions'].items():
        if 'source' not in value:
            continue
        if value['status'] == 'unique':
            mappings[key] = code_record(module, old, new, int(key, 16), int(value['newRva'], 16),
                                        'intrinsic-unique-bounded-signature', value['source'])
        else:
            unresolved.append({'oldRva': key, 'kind': 'code', 'status': value['status'], 'sources': value['source']})
    contextual = {x['oldRva']: x for x in report['contextualFunctionCandidates']}
    for value in list(unresolved):
        candidate = contextual.get(value['oldRva'])
        if not candidate or candidate['status'] != 'single-context-candidate':
            continue
        after = int(candidate['candidates'][0]['newRva'], 16)
        mappings[value['oldRva']] = code_record(module, old, new, int(value['oldRva'], 16), after,
                                               'single-reference-context-plus-byte-signature', value['sources'])
        mappings[value['oldRva']]['evidence'] = candidate['candidates'][0]['evidence']
        unresolved.remove(value)
    for value in report['dataReferences']:
        if value['status'] == 'consistent-reference':
            after = int(value['candidates'][0]['newRva'], 16)
            mappings[value['oldRva']] = data_record(old, new, int(value['oldRva'], 16), after,
                                                   'consistent-paired-reference', value['source'])
            mappings[value['oldRva']]['evidence'] = value['candidates'][0]['references']
        else:
            unresolved.append({'oldRva': value['oldRva'], 'kind': 'data', 'status': value['status'], 'sources': value['source']})
    manual_path = audit_path.with_name('manual-closures.json')
    manual = json.loads(manual_path.read_text(encoding='utf8')) if manual_path.exists() else {'code': [], 'data': []}
    for value in manual['code']:
        mappings[value['oldRva']] = manual_code_record(module, old, new, value)
        unresolved = [x for x in unresolved if x['oldRva'] != value['oldRva']]
    for value in manual['data']:
        mappings[value['oldRva']] = data_record(old, new, int(value['oldRva'], 16), int(value['newRva'], 16),
                                               'explicit-body-proven-array-or-data-binding', value.get('sources'))
        mappings[value['oldRva']]['evidence'] = value['basis']
        unresolved = [x for x in unresolved if x['oldRva'] != value['oldRva']]
    foundation_path = audit_path.with_name('galaxy-foundation-slots.json')
    foundations = json.loads(foundation_path.read_text(encoding='utf8'))
    for value in foundations:
        if not value['normalisedBytesMatch']:
            raise ValueError('An expected foundation slot did not match')
        before, after = int(value['oldFunction'], 16), int(value['newFunction'], 16)
        mappings.setdefault(hex(before), code_record(module, old, new, before, after, 'explicit-foundation-vtable-slot'))
        unresolved = [x for x in unresolved if x['oldRva'] != hex(before)]
    supporting = []
    for name in ('manual-closures.json', 'galaxy-registry-closure.json', 'galaxy-component-bindings.json',
                 'csm-gameplay-string-closure.json', 'dynamics-slot-closure.json',
                 'appearance-manual-triage.json', 'radius-getter-manual-triage.json',
                 'research-switch-table-closures.json', 'non-galaxy-context-closures.json',
                 'commerce-ctor-1e84fc0-closure.json', 'commerce-ctor-1e85ee0-closure.json',
                 'station-send-callers.json'):
        path = audit_path.with_name(name)
        if path.exists():
            supporting.append({'path': str(path), 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()})
    csm = json.loads(audit_path.with_name('csm-gameplay-string-closure.json').read_text(encoding='utf8'))
    registry = json.loads(audit_path.with_name('galaxy-registry-closure.json').read_text(encoding='utf8'))
    registrations = json.loads(audit_path.with_name('galaxy-component-bindings.json').read_text(encoding='utf8'))
    for row in registry['functions']:
        if row['status'] != 'unique':
            raise ValueError('Registry basis remains ambiguous')
        before, after = int(row['oldRva'], 16), int(row['newRva'], 16)
        mappings.setdefault(row['oldRva'], code_record(module, old, new, before, after, 'registry-foundation-byte-match'))
    before, after = int(csm['oldFunction'], 16), int(csm['newFunction'], 16)
    mappings[csm['oldFunction']] = code_record(module, old, new, before, after,
                                             'full-csm-predicate-plus-literal-closure', limit=csm['contiguousSpanLength'])
    published, published_artifacts = published_galaxy_bindings(module, old, new)
    mappings.update(published)
    supporting.extend(published_artifacts)
    profile = {
        'format': 'galaxy-offline-mapping-evidence-v1',
        'runtimeNeutral': True, 'canInstallHooks': False, 'runtimeValidated': False,
        'oldImage': {'path': str(old.path), 'sha256': hashlib.sha256(old.raw).hexdigest()},
        'newImage': {'path': str(new.path), 'sha256': hashlib.sha256(new.raw).hexdigest()},
        'audit': str(audit_path), 'auditSha256': hashlib.sha256(audit_path.read_bytes()).hexdigest(),
        'mappings': list(mappings.values()),
        'foundationSlots': foundations, 'componentBindings': registrations['bindings'],
        'csmStrings': csm['strings'], 'unresolved': unresolved,
        'supportingArtifacts': supporting,
        'requiredManualChecks': [
            'Native calling conventions, trampoline relocation and owning-thread behavior are not validated by this file.',
            'Address candidates do not prove account/server cache readiness, source completeness or new-version live coverage.',
            'Copied object layouts require the original constructor/getter/decoder evidence; byte windows alone do not approve them.',
            'Only a separately reviewed exact-hash native profile may be used; this evidence ledger does not approve installation.'
        ]
    }
    output.write_text(json.dumps(profile, ensure_ascii=False, indent=2), encoding='utf8')
    return profile


def verify(profile_path, old_path=None, new_path=None):
    module = audit_module()
    profile = json.loads(profile_path.read_text(encoding='utf8'))
    if profile.get('runtimeNeutral') is not True or profile.get('canInstallHooks') is not False or profile.get('runtimeValidated') is not False:
        raise ValueError('Evidence file must remain explicitly runtime-neutral')
    audit_bytes = Path(profile['audit']).read_bytes()
    if hashlib.sha256(audit_bytes).hexdigest() != profile['auditSha256']:
        raise ValueError('Underlying reference audit changed')
    report = json.loads(audit_bytes)
    expected = {key: row['newRva'] for key, row in report['functions'].items() if row['status'] == 'unique'}
    expected.update({row['oldRva']: row['candidates'][0]['newRva'] for row in report['dataReferences']
                     if row['status'] == 'consistent-reference'})
    for row in report['contextualFunctionCandidates']:
        if row['status'] == 'single-context-candidate':
            expected.setdefault(row['oldRva'], row['candidates'][0]['newRva'])
    for artifact in profile['supportingArtifacts']:
        path = Path(artifact['path'])
        raw = path.read_bytes()
        if hashlib.sha256(raw).hexdigest() != artifact['sha256']:
            raise ValueError('Supporting evidence changed: ' + path.name)
        if path.name == 'manual-closures.json':
            manual = json.loads(raw)
            for row in manual['code'] + manual['data']:
                expected[row['oldRva']] = row['newRva']
    old = module.Image(old_path or Path(profile['oldImage']['path']))
    new = module.Image(new_path or Path(profile['newImage']['path']))
    for label, image in (('oldImage', old), ('newImage', new)):
        if hashlib.sha256(image.raw).hexdigest() != profile[label]['sha256']:
            raise ValueError(label + ' hash mismatch')
    published, _ = published_galaxy_bindings(module, old, new)
    expected.update({key: row['newRva'] for key, row in published.items()})
    seen = set()
    for row in profile['mappings']:
        if row['oldRva'] in seen:
            raise ValueError('Duplicate old address')
        seen.add(row['oldRva'])
        if row['oldRva'] in expected and expected[row['oldRva']] != row['newRva']:
            raise ValueError('Address disagrees with reference evidence: ' + row['oldRva'])
        before, after = int(row['oldRva'], 16), int(row['newRva'], 16)
        length = row['bytes']
        a, b = old.data(before, length), new.data(after, length)
        if hashlib.sha256(a).hexdigest() != row['oldBytesSha256'] or hashlib.sha256(b).hexdigest() != row['newBytesSha256']:
            raise ValueError('Recorded bytes changed at ' + row['oldRva'])
        if row['kind'] == 'code':
            mask = bytes.fromhex(row['maskHex'])
            if len(mask) != length or any(x != y for x, y, keep in zip(a, b, mask) if keep):
                raise ValueError('Normalized code changed at ' + row['oldRva'])
            if 'internalBranchesVerified' in row and internal_branches(module, old, new, before, after, length) != row['internalBranchesVerified']:
                raise ValueError('Internal branch verification changed')
        elif old.section(before) != new.section(after) or old.section(before) != row['section']:
            raise ValueError('Data section changed')
    for slot in profile['foundationSlots']:
        offset = int(slot['slot'], 16)
        before = int.from_bytes(old.data(int(slot['oldVtable'], 16) + offset, 8), 'little') - old.base
        after = int.from_bytes(new.data(int(slot['newVtable'], 16) + offset, 8), 'little') - new.base
        if before != int(slot['oldFunction'], 16) or after != int(slot['newFunction'], 16):
            raise ValueError('Foundation slot changed: ' + slot['name'])
    for row in profile['csmStrings']:
        a = old.data(int(row['oldRva'], 16), 256).split(b'\0')[0]
        b = new.data(int(row['newRva'], 16), 256).split(b'\0')[0]
        if a != b or a.decode('ascii') != row['oldText'] or b.decode('ascii') != row['newText']:
            raise ValueError('CSM state literal changed')
    for row in profile['componentBindings']:
        for image, prefix in ((old, 'old'), (new, 'new')):
            function = int.from_bytes(image.data(int(row[prefix + 'Vtable'], 16), 8), 'little') - image.base
            if function != int(row[prefix + 'TypeGetter'], 16):
                raise ValueError('Registration type-getter slot changed')
            instruction = image.instructions(function, 6)[0]
            memory = [op for op in instruction.operands if op.type == module.X86_OP_MEM and op.mem.base == module.X86_REG_RIP]
            if instruction.mnemonic != 'mov' or len(memory) != 1 or \
               instruction.address + instruction.size + memory[0].mem.disp - image.base != int(row[prefix + 'TypeIdGlobal'], 16):
                raise ValueError('Registration type-ID global changed')
    inventory = module.source_inventory()
    state_source = Path('native/ManagedPlugin/Navigation.State.cs').read_text(encoding='utf8')
    source_hash = re.search(r'GameHash\s*=\s*"([\da-fA-F]{64})"', state_source).group(1).lower()
    unresolved_uses = []
    if source_hash == profile['oldImage']['sha256']:
        source_version = 'old'
        accounted = {int(x['oldRva'], 16) for x in profile['mappings'] + profile['unresolved']}
    elif source_hash == profile['newImage']['sha256']:
        source_version = 'new'
        accounted = {int(x['newRva'], 16) for x in profile['mappings']}
        for row in profile['unresolved']:
            candidates = report['functions'].get(row['oldRva'], {}).get('candidates', [])
            for candidate in candidates:
                address = int(candidate, 16)
                if address in inventory:
                    accounted.add(address)
                    unresolved_uses.append({'oldRva': row['oldRva'], 'sourceNewRva': candidate,
                                            'status': 'source-uses-unresolved-context-candidate', 'sources': row['sources']})
    else:
        raise ValueError('Current source exact-hash profile is outside these audited images')
    missing = set(inventory) - accounted
    if missing:
        raise ValueError('Current source has uninventoried addresses: ' + ','.join(hex(x) for x in sorted(missing)))
    return {'verifiedMappings': len(profile['mappings']), 'verifiedFoundationSlots': len(profile['foundationSlots']),
            'verifiedRegistrationBindings': len(profile['componentBindings']),
            'verifiedCsmStrings': len(profile['csmStrings']), 'sourceRvasAccounted': len(inventory),
            'sourceImageProfile': source_version, 'sourceUnresolvedUses': unresolved_uses,
            'unresolved': len(profile['unresolved']),
            'runtimeValidated': False, 'canInstallHooks': False}


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--audit', type=Path, default=Path('artifacts/galaxy-client-compatibility/audit-closure.json'))
    parser.add_argument('--output', type=Path, default=Path('artifacts/galaxy-client-compatibility/galaxy-mapping.json'))
    parser.add_argument('--verify', type=Path)
    args = parser.parse_args()
    if args.verify:
        result = verify(args.verify)
    else:
        build(args.audit, args.output)
        result = verify(args.output)
    print(json.dumps(result, indent=2))


if __name__ == '__main__':
    main()