"""Bounded external reads of the current player's build-pinned flight components.
No injection, hooks, native calls, process writes, input or screen access.
The player's identity is obtained from the explicitly started toolbox observer.
"""
import argparse
import ctypes as c
import ctypes.wintypes as w
import hashlib
import json
import math
from pathlib import Path
import struct
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--pid', type=int, required=True)
parser.add_argument('--profile', type=Path, required=True)
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--export-density',type=Path,help='Export only the original immutable Galactic density resource, using VM_READ')
parser.add_argument('--bootstrap-components', action='store_true', help='Read unique type-8 player candidates while diagnosing unavailable pose telemetry; never accepts flight input')
parser.add_argument('--damage',action='store_true',help='Include the build-pinned BasicDamage component using bounded read-only access')
args = parser.parse_args()
profile = json.loads(args.profile.read_text(encoding='utf-8-sig'))
image = Path(profile['image']).resolve(strict=True)
if hashlib.sha256(image.read_bytes()).hexdigest() != profile['sha256']:
raise ValueError('Game build mismatch')
kernel = c.WinDLL('kernel32', use_last_error=True)
kernel.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]
kernel.OpenProcess.restype = w.HANDLE
kernel.CloseHandle.argtypes = [w.HANDLE]
kernel.QueryFullProcessImageNameW.argtypes = [w.HANDLE, w.DWORD, w.LPWSTR, c.POINTER(w.DWORD)]
kernel.ReadProcessMemory.argtypes = [w.HANDLE, c.c_void_p, c.c_void_p, c.c_size_t, c.POINTER(c.c_size_t)]
kernel.CreateToolhelp32Snapshot.argtypes = [w.DWORD, w.DWORD]
kernel.CreateToolhelp32Snapshot.restype = w.HANDLE
class Module(c.Structure):
_fields_ = [('dwSize', w.DWORD), ('th32ModuleID', w.DWORD), ('th32ProcessID', w.DWORD),
('GlblcntUsage', w.DWORD), ('ProccntUsage', w.DWORD), ('modBaseAddr', c.c_void_p),
('modBaseSize', w.DWORD), ('hModule', w.HMODULE), ('szModule', w.WCHAR * 256), ('szExePath', w.WCHAR * 260)]
kernel.Module32FirstW.argtypes = [w.HANDLE, c.POINTER(Module)]
kernel.Module32NextW.argtypes = [w.HANDLE, c.POINTER(Module)]
kernel.WaitNamedPipeW.argtypes = [w.LPCWSTR, w.DWORD]
kernel.CreateFileW.argtypes = [w.LPCWSTR, w.DWORD, w.DWORD, c.c_void_p, w.DWORD, w.DWORD, w.HANDLE]
kernel.CreateFileW.restype = w.HANDLE
kernel.WriteFile.argtypes = [w.HANDLE, c.c_void_p, w.DWORD, c.POINTER(w.DWORD), c.c_void_p]
kernel.ReadFile.argtypes = [w.HANDLE, c.c_void_p, w.DWORD, c.POINTER(w.DWORD), c.c_void_p]
process = kernel.OpenProcess(0x410, False, args.pid)
if not process:
raise c.WinError(c.get_last_error())
def read(at, size):
if at < 0x10000 or not 0 < size <= 8192:
raise ValueError('Read outside bounded component memory')
value = c.create_string_buffer(size)
done = c.c_size_t()
if not kernel.ReadProcessMemory(process, at, value, size, c.byref(done)) or done.value != size:
raise c.WinError(c.get_last_error())
return value.raw
def number(at, kind='<Q'):
return struct.unpack(kind, read(at, struct.calcsize(kind)))[0]
def pipe_call(op):
name = rf'\\.\pipe\XFEEDToolBox.Navigation.{args.pid}'
if not kernel.WaitNamedPipeW(name, 2000):
raise c.WinError(c.get_last_error())
pipe = kernel.CreateFileW(name, 0xC0000000, 0, None, 3, 0, None)
if pipe == c.c_void_p(-1).value:
raise c.WinError(c.get_last_error())
try:
count = w.DWORD()
body = (json.dumps({'op': op}) + '\n').encode()
if not kernel.WriteFile(pipe, body, len(body), c.byref(count), None):
raise c.WinError(c.get_last_error())
data = c.create_string_buffer(65536)
if not kernel.ReadFile(pipe, data, len(data), c.byref(count), None):
raise c.WinError(c.get_last_error())
reply = json.loads(data.raw[:count.value])
if not reply.get('ok'):
raise ValueError(reply.get('error', 'Observer operation failed'))
return reply['result']
finally:
kernel.CloseHandle(pipe)
try:
path = c.create_unicode_buffer(32768)
length = w.DWORD(len(path))
if not kernel.QueryFullProcessImageNameW(process, 0, path, c.byref(length)) or Path(path.value).resolve() != image:
raise ValueError('Wrong selected process image')
modules = kernel.CreateToolhelp32Snapshot(0x18, args.pid)
if modules == c.c_void_p(-1).value:
raise c.WinError(c.get_last_error())
base = image_size = 0
try:
item = Module()
item.dwSize = c.sizeof(Module)
found = kernel.Module32FirstW(modules, c.byref(item))
while found:
if Path(item.szExePath).resolve() == image:
base, image_size = item.modBaseAddr, item.modBaseSize
break
found = kernel.Module32NextW(modules, c.byref(item))
finally:
kernel.CloseHandle(modules)
if not base:
raise ValueError('Game image was not found in selected process')
description = pipe_call('describe')
if description['processId'] != args.pid or description['gameSha256'] != profile['sha256']:
raise ValueError('Observer identity mismatch')
player = {'inGame': description['inGame'], 'entityId': None, 'poseVerified': False} if args.bootstrap_components else pipe_call('read')
if not player['inGame']:
raise ValueError('Player is outside the game')
entity = int(player['entityId']) if player['entityId'] is not None else None
def in_image(at):
return base <= at < base + image_size
def lifecycle(obj):
owner = number(obj + 0x20)
method = number(number(owner) + 0x20)
if not in_image(method):
return None
code = read(method, 12)
reference = 0
if code[:3] == b'\x48\x8d\x81' and code[7] == 0xC3:
reference = owner + struct.unpack_from('<i', code, 3)[0]
elif code[:3] == b'\x48\x8d\x41' and code[4] == 0xC3:
reference = owner + struct.unpack_from('<b', code, 3)[0]
elif code[:3] == b'\x48\x8b\x81' and code[7] == 0xC3:
reference = number(owner + struct.unpack_from('<i', code, 3)[0])
elif code[:3] == b'\x48\x8b\x41' and code[4] == 0xC3:
reference = number(owner + struct.unpack_from('<b', code, 3)[0])
if not reference:
return None
life = number(reference)
return life if number(life + 0xC0, '<I') == 8 and (entity is None or number(life + 0xB0) == entity) else None
components = []
pools=[('heat', profile['heatPool'], 0x190), ('heatManager', profile['heatManagerPool'], 0x228), ('fsd', profile['fsdPool'], 0xEC0)]
if args.damage:
if profile['sha256']!='6891f70a021fa9f3a0e00458c284f48d1226e68ae7fc02ad48081aed47857e93':raise ValueError('Damage layout not known for this build')
pools.append(('damage',0x5FCF500,0x3E8))
pools.append(('docking',0x5FAA2A0,0x750))
for label, rva, size in pools:
pool = base + rva
if number(pool + 0x68) != size or number(pool + 0x90) > 128:
raise ValueError('Component pool size/bound mismatch')
block, visited = number(pool + 0x30), set()
while block and block not in visited and len(visited) < 128:
visited.add(block)
count = number(block + 0x38)
if not 1 <= count <= 128:
raise ValueError('Component block count invalid')
for index in range(count):
obj = block + 0x80 + index * size
try:
if not in_image(number(obj)):
continue
life = lifecycle(obj)
if life is None:
continue
tables = []
for offset in (0, 8, 0x58, 0x60):
table = number(obj + offset)
if not in_image(table):
continue
methods = [number(table + i * 8) for i in range(80)]
tables.append({'objectOffset': hex(offset), 'rva': hex(table - base), 'methods': [hex(m - base) if in_image(m) else None for m in methods]})
components.append({'name': label, 'address': hex(obj), 'lifecycle': hex(life), 'entityId': str(number(life + 0xB0)), 'ownerContext': hex(number(obj + 0x10)), 'tables': tables, 'bytes': read(obj, size).hex()})
except (OSError, ValueError):
continue
block = number(block + 0x20)
if args.bootstrap_components:
candidates=[]
for row in components:
if row['name']!='fsd':continue
try:
fsd_obj=int(row['address'],16);physics_root=number(fsd_obj+0x2A8)-0x58
input_root=number(physics_root+0x3B0)-0x60;controller=number(input_root+0x2F8)
if not in_image(number(controller)) or number(number(controller)+0x158)!=base+0x25E5DD0:continue
local_player=number(input_root+0x2E8);local_table=number(local_player);controlled_getter=number(local_table+0x160)
local_source={'address':hex(local_player),'tableRva':hex(local_table-base),'getControlledRva':hex(controlled_getter-base),
'getterBytes':read(controlled_getter,32).hex(),'header':read(local_player,0x800).hex()}
if read(controlled_getter,8)==bytes.fromhex('488d8138010000c3'):
pilot_life=number(local_player+0x138);pilot_root=number(pilot_life+0xC8)-0x30
pilot_type=number(base+0x5F05BA4,'<I');pilot_meta=number(pilot_root+0x10)
pilot_slot=number(number(pilot_meta+0x30)+pilot_type*16)
if pilot_slot<number(pilot_meta+0x18):
pilot_interface=number(number(pilot_root+0x18)+pilot_slot*8)
ship_getter=number(number(pilot_interface)+0x68)
local_source['pilot']={'life':hex(pilot_life),'root':hex(pilot_root),'typeId':pilot_type,'denseSlot':pilot_slot,
'interface':hex(pilot_interface),'tableRva':hex(number(pilot_interface)-base),'getShipRva':hex(ship_getter-base),
'getterBytes':read(ship_getter,256).hex(),'header':read(pilot_interface,0x400).hex()}
actor_owner=pilot_interface-0x58;actor_count=number(actor_owner+0x970);actor_array=number(actor_owner+0x978)
if not 0<actor_count<=32:raise ValueError('Local player actor list exceeds bound')
actors=[]
for actor_index in range(actor_count):
actor=actor_array+actor_index*0x1C0;actor_life=number(actor+0x1B0)
actors.append({'kind':number(actor+0x190,'<B'),'ticket':number(actor+0x1B8,'<i'),
'life':hex(actor_life),'entityId':str(number(actor_life+0xB0)) if actor_life else None,
'lifeType':number(actor_life+0xC0,'<I') if actor_life else None})
eligible=[row for row in actors if row['ticket']!=-1]
chosen=next((row for kind in (2,0,3) for row in eligible if row['kind']==kind),None)
local_source['pilot']['actors']=actors
local_source['pilot']['selectedPlayerShip']=chosen
local_source['pilot']['matchesCandidate']=chosen is not None and chosen['life']==row['lifecycle']
candidates.append({'entityId':row['entityId'],'lifecycle':row['lifecycle'],'fsd':row['address'],
'physics':hex(physics_root),'inputRoot':hex(input_root),'controller':hex(controller),'controllerTableRva':hex(number(controller)-base),
'controllerHeader':read(controller,0xC00).hex(),'inputHeader':read(input_root,0x400).hex(),
'physicsHeader':read(physics_root,0xB00).hex(),'localPlayerSource':local_source})
except (OSError,ValueError):continue
report={'pid':args.pid,'base':hex(base),'gameSha256':profile['sha256'],'unclassifiedType8Candidates':candidates,
'components':components,'gameInputsSent':0,'screenUse':False,'processAccess':'QUERY_INFORMATION | VM_READ'}
with args.output.open('x',encoding='utf-8') as output:json.dump(report,output,indent=2)
print(json.dumps({'unclassifiedCandidates':[{k:v for k,v in row.items() if not k.endswith('Header')} for row in candidates]}))
return
primary_components=[row for row in components if row['name'] in ('heat','heatManager','fsd')]
if len(primary_components) != 3 or len({row['ownerContext'] for row in components}) != 1 or len({row['entityId'] for row in components}) != 1:
raise ValueError('The three unique player components did not match')
life = int(components[0]['lifecycle'], 16)
root = number(life + 0xC8) - 0x30
getter = base + 0x8AE6F0
code = read(getter, 12)
if code[:2] != b'\x8b\x05' or code[6:12] != b'\x89\x02\x48\x8b\xc2\xc3':
raise ValueError('Pose ID getter signature mismatch')
type_id = number(getter + 6 + struct.unpack_from('<i', code, 2)[0], '<I')
if not 0 < type_id < 100000:
raise ValueError('Pose type ID outside diagnostic bound')
metadata = number(root + 0x10)
slots = number(metadata + 0x30)
slot = number(slots + type_id * 16)
if not 0 <= slot < 512:
raise ValueError('Resolved pose slot outside diagnostic bound')
pose = number(number(root + 0x18) + slot * 8)
pose_getter = number(number(pose) + 0x28)
pose_lookup = {'root': hex(root), 'metadata': hex(metadata), 'poseTypeId': type_id, 'denseSlotIndex': slot,
'poseAddress': hex(pose), 'getterRva': hex(pose_getter - base), 'getterMatches': pose_getter == base + 0x3CBC080,
'metadataHeader': read(metadata, 128).hex(), 'rootHeader': read(root, 128).hex()}
physics = []
component_count = number(metadata + 0x18)
if not 1 <= component_count <= 1024:
raise ValueError('Dense component count outside diagnostic bound')
array = number(root + 0x18)
if args.damage:
interfaces=[]
for dense_index in range(component_count):
try:
candidate=number(array+dense_index*8);table=number(candidate)
if not in_image(table):continue
methods=[number(table+i*8) for i in range(80)]
interfaces.append({'index':dense_index,'address':hex(candidate),'tableRva':hex(table-base),'header':read(candidate,128).hex(),
'methods':[hex(m-base) if in_image(m) else None for m in methods]})
except (OSError,ValueError):continue
player['readOnlyInterfaces']=interfaces
damage_type=number(base+0x5F2F254,'<I')
damage_slot=number(number(metadata+0x30)+damage_type*16)
if damage_slot<component_count:
damage=number(array+damage_slot*8);encoded=number(damage+0x178)
property_object=number(damage-0x58+0x120);property_table=number(property_object)
import math
value=struct.unpack('<f',struct.pack('<I',(encoded>>32)^(encoded&0xffffffff)))[0]
player['readOnlyDamage']={'interface':hex(damage),'tableRva':hex(number(damage)-base),'typeId':damage_type,'denseSlot':damage_slot,
'numerator':value if math.isfinite(value) else None,'property':hex(property_object),'propertyTableRva':hex(property_table-base),
'propertyGetterRva':hex(number(property_table+0x18)-base),'propertyHeader':read(property_object,128).hex(),
'header':read(damage-0x58,0x3e8).hex()}
inspected = set()
for index in range(component_count):
candidate = number(array + index * 8)
for adjustment in (0, 8, 0x58, 0x60):
obj = candidate - adjustment
if obj in inspected:
continue
inspected.add(obj)
try:
if not in_image(number(obj)) or lifecycle(obj) != life:
continue
input_root = number(obj + 0x3B0) - 0x60
if not in_image(number(input_root)):
continue
controller = number(input_root + 0x2F8)
if number(number(controller) + 0x158) != base + 0x25E5DD0:
continue
physics.append({'address': hex(obj), 'denseSlot': index, 'inputRoot': hex(input_root),
'inputTableRva': hex(number(input_root) - base), 'controller': hex(controller),
'controllerTableRva': hex(number(controller) - base), 'inputBytes': read(input_root, 0x580).hex(),
'physicsBytes': read(obj, 0xB00).hex(), 'controllerMethods': [hex(number(number(controller) + i * 8) - base) for i in range(64)]})
except (OSError, ValueError):
continue
report = {'pid': args.pid, 'gameSha256': profile['sha256'], 'base': hex(base), 'bootstrapCandidatesOnly': args.bootstrap_components,
'player': player, 'poseLookup': pose_lookup, 'components': components, 'physics': physics,
'poseBytes': read(pose, 0x480).hex(), 'gameInputsSent': 0, 'screenUse': False, 'processAccess': 'QUERY_INFORMATION | VM_READ'}
maintenance_pool=base+0x5F85A60
maintenance=dict(stride=number(maintenance_pool+0x68),count=number(maintenance_pool+0x90),objects=[])
if maintenance['stride']==0x440 and maintenance['count']<=128:
block=number(maintenance_pool+0x30);visited=set()
while block and block not in visited and len(visited)<128:
visited.add(block);count=number(block+0x38)
if not 1<=count<=128:raise ValueError('Maintenance pool block exceeds bounds')
for index in range(count):
obj=block+0x80+index*0x440
if number(obj) not in (base+0x52409D0,base+0x5240888):continue
table=number(obj+0x58)
maintenance['objects'].append(dict(address=hex(obj),interface=hex(obj+0x58),table=hex(table-base),header=read(obj,0x440).hex(),
methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(40)]]))
block=number(block+0x20)
report['maintenancePool']=maintenance
fsd_address = int(next(row for row in components if row['name'] == 'fsd')['address'], 16)
sources = []
for label, offset in [('worldTransform', 0x320), ('navigationTarget', 0x338), ('spaceBodies', 0x6A0), ('actualFlightControl',0x2A8), ('heatInput',0x2F0),
('localFlightState',0x3F8),('cruiseTargetService',0x470),('playerController',0x6B0),('fsdEligibility',0x6D0)]:
interface = number(fsd_address + offset)
table = number(interface)
if not in_image(table):
raise ValueError('FSD source interface table outside selected image')
methods = [number(table + i*8) for i in range(112)]
sources.append({'name': label, 'address': hex(interface), 'tableRva': hex(table-base),
'methods': [hex(m-base) if in_image(m) else None for m in methods], 'header': read(interface, 512).hex()})
report['fsdSources'] = sources
# Original galaxy route interfaces, looked up through registered component
# IDs. This records pointers and method tables only; no native invocation.
route_roots = [('ship', root)]
if len(physics) == 1:
input_root = int(physics[0]['inputRoot'], 16)
local_player = number(input_root + 0x2E8)
controlled_getter = number(number(local_player) + 0x160)
if read(controlled_getter, 8) == bytes.fromhex('488d8138010000c3'):
pilot_life = number(local_player + 0x138)
route_roots.append(('pilot', number(pilot_life + 0xC8) - 0x30))
route_bindings = []
route_components = []
for root_name, route_root in route_roots:
route_meta = number(route_root + 0x10)
route_slots = number(route_meta + 0x30)
route_array = number(route_root + 0x18)
route_count = number(route_meta + 0x18)
if not 1 <= route_count <= 1024:
raise ValueError('Route component registry exceeds its bound')
for index in range(route_count):
component = number(route_array + index * 8)
try:
table = number(component)
if in_image(table):
item=dict(owner=root_name, denseSlot=index,
address=hex(component), tableRva=hex(table-base), header=read(component, 0x80).hex())
if args.damage:
item['methods']=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(80)]]
header=read(component-0x58,0x458)
for offset in range(0,len(header)-7,8):
if struct.unpack_from('<Q',header,offset)[0]==base+0x5322218:
ui=component-0x58+offset;assist=number(ui+0x258);at=number(assist)
if in_image(at):
item['supercruiseUi']=dict(address=hex(ui),assist=hex(assist),assistTable=hex(at-base),
assistHeader=read(assist,0x600).hex(),assistMethods=[hex(v-base) if in_image(v) else None for v in [number(at+i*8) for i in range(32)]])
route_components.append(item)
except (OSError, ValueError):
continue
for name, id_rva in [('route',0x5F05DD0), ('routeController',0x5F59D90),
('mapRouteRequest',0x5F05DD4), ('mapRoutePlotter',0x5F05DCC),
('world',0x5F05D3C), ('targets',0x5F44D04), ('galaxy',0x5F05CC8),
('galaxyCatalogue',0x5F05C1C),('jumpConfiguration',0x5F05CB4),('supercruiseAssist',0x5F53170),
('cruiseControlGeometry',0x5F53174),('cruiseArrival',0x5F34640),('cruiseNavigationWorld',0x5F34660),('cruiseInputTransform',0x5F2F27C),
('flightModeInterface',0x5F2F260),('dockingMotionInterface',0x5F44D48),('dockingGuideInterface',0x5F44CF0),
('dockingAssist',0x5F53178),('launchAssist',0x5F5317C),('shipActivityController',0x5F44C98),
('cargoInventory',0x5F2F25C),('fuelInventory',0x5F45128),('maintenanceOperations',0x5F7E800),
('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC)]:
type_id = number(base + id_rva, '<I')
if not 0 < type_id < 100000:
continue
index = number(route_slots + type_id * 16)
if index >= route_count:
continue
interface = number(route_array + index * 8)
if interface < 0x10000:
continue
table = number(interface)
if not in_image(table):
continue
binding_row=dict(owner=root_name, name=name, root=hex(route_root),
typeId=type_id, denseSlot=index, address=hex(interface), tableRva=hex(table-base),
methods=[hex(method-base) if in_image(method) else None
for method in [number(table+i*8) for i in range(64)]], header=read(interface, 0x700).hex())
if name in ('supercruiseAssist','cruiseControlGeometry'):
parameters=number(interface-8)
if parameters:binding_row['parameters']=dict(address=hex(parameters),data=read(parameters,0x100).hex())
if name=='launchAssist':
bindings=number(interface+0x230)
if bindings:
table=number(bindings)
binding_row['inputBindings']=dict(address=hex(bindings),table=hex(table-base),
getBinding=hex(number(table+0x40)-base),header=read(bindings,0x80).hex())
brain=number(interface+0x1D0)
if brain and in_image(number(brain)):
getter=number(number(brain)+0xA0)
if in_image(getter):binding_row['brain']=dict(address=hex(brain),table=hex(number(brain)-base),
dutyGetter=hex(getter-base),dutyGetterCode=read(getter,64).hex(),header=read(brain,0x200).hex())
if read(getter,8)==bytes.fromhex('488b81e8010000c3'):
duty=number(brain+0x1E8)
if duty and in_image(number(duty)):
table=number(duty)
binding_row['brain']['duty']=dict(address=hex(duty),table=hex(table-base),typeGetter=hex(number(table+0x20)-base))
if table==base+0x5156130:
binding_row['brain']['duty']['header']=read(duty,0x378).hex()
refs=[]
for label,offset in [('station',0x300),('pad',0x318),('approach',0x330)]:
cached=number(duty+offset+0x10)
if cached>0x10000 and in_image(number(cached)):
cached_table=number(cached)
refs.append(dict(name=label,address=hex(cached),table=hex(cached_table-base),
header=read(cached,0x200).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(cached_table+i*8) for i in range(48)]]))
binding_row['brain']['duty']['typedComponents']=refs
control=number(interface+0x240)
if control and in_image(number(control)):
table=number(control)
binding_row['desiredInput']=dict(address=hex(control),table=hex(table-base),
setter=hex(number(table+0x158)-base),clearer=hex(number(table+0x150)-base),throttle=number(control+0x680,'<f'))
if name=='supercruiseAssist':
main_table=number(interface-0x58)
if in_image(main_table):binding_row['mainTable']=hex(main_table-base);binding_row['mainMethods']=[hex(v-base) if in_image(v) else None for v in [number(main_table+i*8) for i in range(32)]]
internals=[]
for label,offset in [('approachGeometry',0x90),('flightController',0xF0),('inputs',0x198),('throttleBinding',0x258)]:
source=number(interface-0x58+offset)
if not source:continue
vt=number(source)
internals.append(dict(name=label,address=hex(source),table=hex(vt-base) if in_image(vt) else None,header=read(source,0x300).hex(),
methods=[hex(v-base) if in_image(v) else None for v in [number(vt+i*8) for i in range(64)]] if in_image(vt) else []))
binding_row['internals']=internals
if name=='jumpConfiguration' and number(table+0x130)==base+0x2C9A7A0:
holder=number(interface+0x78)
if holder and number(holder+0x10):
owner_data=number(holder+0x18);configuration=number(owner_data+0x270)
if configuration:binding_row['jumpParameters']=dict(address=hex(configuration),data=read(configuration,64).hex(),floats=list(struct.unpack('<14f',read(configuration,56))))
route_bindings.append(binding_row)
report['routeBindings'] = route_bindings
global_lookup=[]
for binding in route_bindings:
if binding['owner']!='pilot' or binding['name']!='route':continue
source=int(binding['address'],16)-0x58+8
for step,slot in enumerate([0,0x20]):
method=number(number(source)+slot); code=read(method,16)
row=dict(step=step,address=hex(source),method=hex(method-base),code=code.hex())
if code[:3]==b'\x48\x8b\x41' and code[4]==0xc3:source=number(source+struct.unpack_from('<b',code,3)[0])
elif code[:3]==b'\x48\x8b\x81' and code[7]==0xc3:source=number(source+struct.unpack_from('<i',code,3)[0])
elif code[:3]==b'\x48\x8d\x41' and code[4]==0xc3:source+=struct.unpack_from('<b',code,3)[0]
elif code[:3]==b'\x48\x8d\x81' and code[7]==0xc3:source+=struct.unpack_from('<i',code,3)[0]
else:global_lookup.append(row);break
row['result']=hex(source);global_lookup.append(row)
if source>0x10000 and in_image(number(source)):
global_lookup.append(dict(source=hex(source),contextId=number(base+0x5F05AB8,'<I'),
table=hex(number(source)-base),methods=[hex(number(number(source)+i*8)-base) for i in range(8)],header=read(source,128).hex()))
count=number(source+0x28); buckets=number(source+0x20);context_id=number(base+0x5F05AB8,'<I')
if 0<count<=1024 and number(number(source))==base+0x529a30:
for bucket in range(count):
sentinel=buckets+bucket*8; node=number(sentinel); seen=set()
while node and node!=sentinel and node not in seen and len(seen)<1024:
seen.add(node)
for label,id_rva in [('stationMaintenance',0x5F05ACC),('stationServices',0x5F05B94),('economyCatalog',0x5F05CD0),('commanderServices',0x5F05BAC),('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC)]:
if number(node+8,'<I')==number(base+id_rva,'<I'):
service=number(node+16);table=number(service)
if in_image(table):global_lookup.append(dict(domainService=label,typeId=number(node+8,'<I'),address=hex(service),table=hex(table-base),header=read(service,0x600).hex(),methods=[hex(number(table+i*8)-base) if in_image(number(table+i*8)) else None for i in range(80)]))
if number(node+8,'<I')==context_id:
ctx=number(node+16); method=number(number(ctx)+0x18)
code=read(method,16)
row=dict(context=hex(ctx),method=hex(method-base),code=code.hex(),header=read(ctx,128).hex())
if code[:3]==b'\x48\x8b\x41' and code[4]==0xc3:row['value']=hex(number(ctx+struct.unpack_from('<b',code,3)[0]))
elif code[:3]==b'\x48\x8b\x81' and code[7]==0xc3:row['value']=hex(number(ctx+struct.unpack_from('<i',code,3)[0]))
global_lookup.append(row)
if method==base+0x1f3150:
weak=number(ctx+0x568); entity_component=number(weak+0xC8) if weak else 0
if entity_component:
global_root=entity_component-0x30; meta=number(global_root+0x10);slots=number(meta+0x30);array=number(global_root+0x18);n=number(meta+0x18)
if not 1<=n<=2048:raise ValueError('Global component registry outside bound')
row['globalRoot']=hex(global_root); row['globalCount']=n
display_matches=[]
for component_index in range(n):
candidate=number(array+component_index*8)
if candidate<0x10000:continue
try:
header=read(candidate,2048);at=header.find(struct.pack('<I',0xd72f6ba2))
if at>=0:display_matches.append(dict(slot=component_index,address=hex(candidate),table=hex(number(candidate)-base),gdmOffset=hex(at),header=header.hex()))
except (OSError,ValueError):continue
row['galaxyDisplayComponents']=display_matches
global_catalogue=[]
for label,rva in [('galaxy',0x5F05C1C),('galaxyNames',0x5F05C20),('catalogue',0x5F05C24),('galaxyService',0x5F05CC8),('stationMaintenance',0x5F05ACC),('stationServices',0x5F05B94),('economyCatalog',0x5F05CD0),('commanderServices',0x5F05BAC),('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC),('commodityDefinitions',0x5F05C34)]:
kind=number(base+rva,'<I');index=number(slots+kind*16)
if index<n:
obj=number(array+index*8)
if obj<0x10000:continue
table=number(obj)
if not in_image(table):continue
entry=dict(name=label,typeId=kind,address=hex(obj),table=hex(table-base),header=read(obj,1024).hex(),methods=[hex(number(table+i*8)-base) for i in range(80)])
if label=='galaxy':
origin=number(obj+0x188)
if origin>0x10000:entry['origin']=list(struct.unpack('<iii',read(origin,12)))
density=number(obj+0x100)
if density>0x10000:
grid=number(density+0x78)
entry['density']=dict(address=hex(density),header=read(density,144).hex(),grid=hex(grid))
entry['density']['mips']=[]
for mip in range(8):
raster=number(density+0x38+mip*8)
if raster>0x10000:entry['density']['mips'].append(dict(level=mip,address=hex(raster),header=read(raster,32).hex(),width=number(raster,'<I'),height=number(raster+4,'<I')))
if grid>0x10000:entry['density'].update(gridHeader=read(grid,32).hex(),width=number(grid,'<I'),height=number(grid+4,'<I'))
if args.export_density and grid>0x10000:
grid=number(density+0x38)
width=number(grid,'<I');height=number(grid+4,'<I');rows=number(grid+0x10)
if not 16<=width<=8192 or not 16<=height<=8192 or width*height>16777216:raise ValueError('Galactic raster exceeds bound')
pixels=bytearray(width*height*4)
for y in range(height):
density_row=number(rows+y*8);x=0;blocks=0
while x<width and blocks<=width:
blocks+=1;header=number(density_row,'<I');density_row+=4;length=header&0x7fffffff
if not 0<length<=width-x:raise ValueError('Invalid Galactic density RLE')
if header&0x80000000:
value=read(density_row,4);density_row+=4;chunk=value*length
else:
chunk=read(density_row,length*4);density_row+=length*4
start=(y*width+x)*4;pixels[start:start+length*4]=chunk;x+=length
if x!=width:raise ValueError('Incomplete Galactic density row')
sx=number(density+4,'<I');sz=number(density+0x10,'<I')
geometry=(entry['origin'][0]/32,entry['origin'][2]/32,width*2147483648*10/sx,height*2147483648*10/sz)
args.export_density.parent.mkdir(parents=True,exist_ok=True)
import array as density_array
original=memoryview(pixels).cast('I');maximum=max(original)
normalized=density_array.array('f',(value/maximum for value in original)).tobytes()
args.export_density.write_bytes(b'XFEGDM2\0'+struct.pack('<II4d',width,height,*geometry)+normalized)
entry['density']['exported']=dict(file=str(args.export_density),dimensions=[width,height],geometry=geometry,maximum=maximum,sha256=hashlib.sha256(pixels).hexdigest())
cached=[]
for wanted in [int(player['geometry']['worldAddress'])&0x7fffffffffffff,182359951707,10477373803]:
node=number(obj+0x370);depth=0
while node and depth<64:
depth+=1;key=number(node+0x20)
if key==wanted:
galaxy_object=number(node+0x28)
if galaxy_object:
compact=number(galaxy_object+0x38)
if compact:cached.append(dict(system=str(wanted),object=hex(galaxy_object),compact=hex(compact),data=read(compact,104).hex()))
break
node=number(node+(8 if key<wanted else 0))
entry['cachedRecords']=cached
engine=number(obj+0x228)
if engine>0x10000:
vt=number(engine)
if in_image(vt):entry['generator']=dict(address=hex(engine),table=hex(vt-base),header=read(engine,512).hex(),methods=[hex(number(vt+i*8)-base) for i in range(24)])
global_catalogue.append(entry)
row['catalogues']=global_catalogue
node=number(node)
report['globalLookup']=global_lookup
report['registeredRouteComponents'] = route_components
route_lists=[]
for binding in route_bindings:
if binding['owner']!='pilot' or binding['name']!='route':
continue
sentinel=int(binding['address'],16)+0x110
directions=[]
for direction in (0,8):
node=number(sentinel+direction);addresses=[];seen=set()
while node!=sentinel and node and len(seen)<25000:
if node in seen:raise ValueError('Unexpected cycle in game route list')
seen.add(node);addresses.append(str(number(node+0x10)));node=number(node+direction)
directions.append(dict(direction=direction,terminated=node==sentinel,addresses=addresses))
route_lists.append(dict(sentinel=hex(sentinel),directions=directions))
report['originalRouteLists']=route_lists
# D build's FuelScoopModuleComponent registration: native size 0x210 and pool RVA.
# Keep optional diagnostics separate from the three required ship components.
extra=[];unassociated=[]
scoop_pool=base+0x5FA81A0
report['scoopPoolMetadata']={'rva':'0x5fa81a0','stride':number(scoop_pool+0x68),'count':number(scoop_pool+0x90),'header':read(scoop_pool,0xC0).hex()}
if number(scoop_pool+0x68)==0x210 and number(scoop_pool+0x90)<=128:
block=number(scoop_pool+0x30);visited=set()
while block and block not in visited and len(visited)<128:
visited.add(block);count=number(block+0x38)
if not 1<=count<=128:raise ValueError('Fuel scoop block count exceeds bound')
for index in range(count):
obj=block+0x80+index*0x210
table=number(obj)
if not in_image(table):continue
methods=[number(table+i*8) for i in range(80)]
row={'name':'fuelScoop','address':hex(obj),'ownerContext':hex(number(obj+0x10)),
'tableRva':hex(table-base),'methods':[hex(m-base) if in_image(m) else None for m in methods],'header':read(obj,0x210).hex()}
if number(obj+0x10)==int(components[0]['ownerContext'],16):extra.append(row)
elif len(unassociated)<16:
owner=number(obj+0x20)
row['ownerHeader']=read(owner,0x100).hex()
row['ownerAddress']=hex(owner)
row['matchingPlayerLifecycle']=lifecycle(obj) is not None
unassociated.append(row)
block=number(block+0x20)
report['extraComponents']=extra
report['unassociatedFuelScoops']=unassociated
manager = int(next(row for row in components if row['name'] == 'heatManager')['address'], 16)
modifier_count, modifier_array = number(manager + 0x160), number(manager + 0x168)
if not 0 <= modifier_count <= 32 or modifier_count and not modifier_array:
raise ValueError('Heat modifier list exceeds the bounded observation range')
modifiers = []
for index in range(modifier_count):
modifier = number(modifier_array + index*8)
table = number(modifier)
if not in_image(table):
raise ValueError('Heat modifier table outside the selected game image')
methods = [number(table + i*8) for i in range(12)]
modifiers.append({'index':index, 'address':hex(modifier), 'tableRva':hex(table-base),
'methods':[hex(method-base) if in_image(method) else None for method in methods],
'header':read(modifier, 512).hex()})
if number(manager + 0x160) != modifier_count or number(manager + 0x168) != modifier_array:
raise ValueError('Heat modifier list changed while reading')
report['heatModifiers'] = modifiers
navigation = int(next(row for row in sources if row['name'] == 'navigationTarget')['address'], 16)
nav_count, nav_capacity, nav_storage = number(navigation + 0x2C0), number(navigation + 0x2C8), number(navigation + 0x2D0)
if nav_count > 1024 or nav_capacity > 16384 or nav_count > nav_capacity or nav_count and not nav_storage:
raise ValueError('Navigation record collection exceeds the bounded observation range')
records = []
if nav_count:
if nav_capacity == 0 or nav_capacity & (nav_capacity-1):
raise ValueError('Navigation record hash capacity is not a power of two')
occupied = read(nav_storage, (nav_capacity+7)//8)
entries = nav_storage + (((nav_capacity>>3)+7)&~7)
def name_at(record, offset):
try:
text = number(record + offset)
if not text: return ''
length = number(text)
if not 0 < length <= 128: return ''
return read(text+0x14, length).decode('utf-8').replace('\r',' ').replace('\n',' ')
except (OSError, ValueError, UnicodeError): return ''
for index in range(nav_capacity):
if not occupied[index>>3] & (1<<(index&7)): continue
entry=entries+index*16
record=number(entry+8)
if not record: continue
records.append({'key':number(entry,'<i'),'address':hex(record),
'worldAddress':str(number(record+0x20)),
'name':name_at(record,0x140),'secondaryName':name_at(record,0x148),
'kind':number(record+0x184,'<I'),'type':number(record+0x188,'<I'),
'header':read(record,0x1E0).hex()})
if len(records) != nav_count:
raise ValueError('Navigation records changed or had null entries during observation')
if number(navigation+0x2C0)!=nav_count or number(navigation+0x2C8)!=nav_capacity or number(navigation+0x2D0)!=nav_storage:
raise ValueError('Navigation records changed during observation')
report['navigationRecords']={'count':nav_count,'selectedKey':number(navigation+0x38C,'<i'),'records':records}
if args.damage:
# The original life-reference lookup (0x62D120) uses this world's
# entity directory. Check the own-ship key/value before trusting it.
player_life=int(components[0]['lifecycle'],16)
context=number(player_life+0x38)
directory=dict(context=hex(context),header=read(context+0x140,0x30).hex())
count,capacity,storage=number(context+0x148),number(context+0x150),number(context+0x158)
directory.update(count=count,capacity=capacity,storage=hex(storage))
if count<=capacity and 8<=capacity<=65536 and not capacity&(capacity-1) and storage:
occupied=read(storage,(capacity+7)//8);entries=storage+(((capacity>>3)+7)&~7)
markets={struct.unpack_from('<I',bytes.fromhex(r['header']),0x130)[0] for r in records if r['kind'] in (6,12,14,18,28,35)}
matches=[];own=False
for i in range(capacity):
if not occupied[i>>3]&(1<<(i&7)):continue
at=entries+i*16;key,value=number(at),number(at+8)
if key==number(player_life+0xB0):own=value==player_life
if key in markets:matches.append(dict(key=str(key),life=hex(value),type=number(value+0xC0,'<I'),entityId=str(number(value+0xB0))))
directory.update(ownShipMatches=own,stationMatches=matches)
report['entityDirectory']=directory
# Independently reproduce only the immutable affine transforms used
# by the ATC distance guard; no original getter or request is invoked.
def registered(root,type_id):
meta=number(root+0x10);count=number(meta+0x18);slots=number(meta+0x30)
slot=number(slots+type_id*16)
return number(number(root+0x18)+slot*8) if 0<count<=1024 and slot<count else 0
def body_hash(key):
mask=(1<<64)-1;value=(key*0x40000+(~key&mask))&mask
value=((value>>31^value)*21)&mask;value=((value>>11^value)*65)&mask
return ((value>>22)^value)&0xffffffff
def body_entry(table,capacity,key,payload,system=False):
if not table or not 0<capacity<=65536:return 0
sentinel=table+8*(body_hash(key)%capacity);node=number(sentinel)
for _ in range(4096):
if not node or node==sentinel:return 0
if number(node+8)==key and (not system or number(node+16,'<B')==0):return number(node+payload)
node=number(node)
return 0
def affine(at):return [struct.unpack('<3d',read(at+i,24)) for i in (0,0x20,0x40,0x60)]
def rotate(v,m):return tuple(sum(v[j]*m[j][i] for j in range(3)) for i in range(3))
def compose(a,b):return [rotate(v,b) for v in a[:3]]+[tuple(x+y for x,y in zip(rotate(a[3],b),b[3]))]
service=int(next(r['address'] for r in sources if r['name']=='spaceBodies'),16);owner=service-0x58
ship_pose=int(pose_lookup['poseAddress'],16);ship_world=number(ship_pose+0xE8)
resource=body_entry(number(owner+0x278),number(owner+0x280),ship_world&0x007fffffffffffff,24,True)
def pose_position(pose):
table=number(pose)
if table==base+0x553DF90 and number(table+0x60)==base+0x3CBC0E0:universal=pose+0xA8
elif number(table+0x60)==base+0x2D9C2C0:universal=pose+0xE8
else:raise ValueError('Unverified original universal pose implementation')
world=number(universal);body=body_entry(number(resource+0x140),number(resource+0x148),world,16)
if not body or number(body+0x378)!=resource:raise ValueError('ATC pose body/resource mismatch')
index=number(body+0x384,'<I');transforms=number(resource+0x388)
if index>=4096:raise ValueError('ATC pose transform index exceeds bounds')
result=affine(universal+0x10)
if number(universal+0x90,'<B'):result=compose(result,affine(transforms+index*0x180+0x80))
return world,compose(result,affine(transforms+index*0x180+0x100))[3]
checks=[]
for item in directory.get('stationMatches',[]):
check=dict(item)
try:
station_root=number(int(item['life'],16)+0xC8)-0x30
station_bound=[]
for label,id_rva in [('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC),('marketServicesData',0x5F44D10),('stationData',0x5F05ACC)]:
bound=registered(station_root,number(base+id_rva,'<I'))
if bound:
table=number(bound)
station_bound.append(dict(name=label,address=hex(bound),table=hex(table-base),header=read(bound,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(32)]]))
check['boundServices']=station_bound
station_service=registered(station_root,number(base+0x5F771D0,'<I'))
if station_service:
service_table=number(station_service)
check['stationServices']=dict(address=hex(station_service),table=hex(service_table-base),header=read(station_service,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(service_table+i*8) for i in range(48)]])
station_pose=registered(number(int(item['life'],16)+0xC8)-0x30,pose_lookup['poseTypeId'])
station_table=number(station_pose)
check.update(stationPose=hex(station_pose),table=hex(station_table-base),resource=hex(resource),
stationWorld=str(number(station_pose+0xE8)),shipWorld=str(ship_world),poseHeader=read(station_pose,0x200).hex(),
methods=[hex(number(station_table+i*8)-base) for i in range(14)])
world,position=pose_position(station_pose);_,ship_position=pose_position(ship_pose)
check.update(stationWorld=str(world),shipWorld=str(ship_world),sameSystem=(world&0x007fffffffffffff)==(ship_world&0x007fffffffffffff),
actualDistance=math.dist(position,ship_position),stationPosition=position,shipPosition=ship_position)
except (OSError,ValueError) as error:check['failure']=str(error)
checks.append(check)
report['atcStationGeometry']=checks
service=int(next(r['address'] for r in sources if r['name']=='cruiseTargetService'),16)
targets=[]
for offset in (0x2E0,0x2F0,0x300):
candidate=number(service+offset);item=dict(offset=hex(offset),life=hex(candidate))
try:
if candidate and number(candidate+0xC0,'<I')>4:
item.update(entityId=str(number(candidate+0xB0)),type=number(candidate+0xC0,'<I'))
target_root=number(candidate+0xC8)-0x30;meta=number(target_root+0x10);slots=number(meta+0x30);dense_count=number(meta+0x18)
kind=number(base+0x5F976B0,'<I');slot=number(slots+kind*16)
if 1<=dense_count<=1024 and slot<dense_count:
component=number(number(target_root+0x18)+slot*8);table=number(component)
if in_image(table):item['dropComponent']=dict(address=hex(component),table=hex(table-base),header=read(component,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(32)]])
except (OSError,ValueError):pass
targets.append(item)
report['cruiseTargetReferences']=dict(mode=number(service+0x2F8,'<I'),targets=targets)
if pose_lookup.get('root'):
root=int(pose_lookup['root'],16);meta=number(root+0x10);dense=number(meta+0x18);slots=number(meta+0x30)
slot=number(slots+number(base+0x5F2F274,'<I')*16)
if 0<dense<=1024 and slot<dense:
loadout=number(number(root+0x18)+slot*8);table=number(loadout)
report['originalAmmoLoadout']={'address':hex(loadout),'table':hex(table-base),'header':read(loadout,0x400).hex(),
'methods':[hex(number(table+i*8)-base) for i in range(40)]}
ammo=[];count=number(loadout+0x1D0);rows=number(loadout+0x1D8)
if count<=128:
for index in range(count):
life=number(rows+index*0x30)
if not life:continue
try:
module_root=number(life+0xC8)-0x30;module_meta=number(module_root+0x10);n=number(module_meta+0x18)
module_slots=number(module_meta+0x30);components=[]
for label,rva in [('clip',0x5F44D80),('hopper',0x5F53A7C)]:
kind=number(base+rva,'<I');dense_index=number(module_slots+kind*16)
if not 0<n<=1024 or dense_index>=n:continue
component=number(number(module_root+0x18)+dense_index*8);vt=number(component)
if in_image(vt):components.append({'kind':label,'address':hex(component),'table':hex(vt-base),'header':read(component,0x100).hex(),
'methods':[hex(number(vt+j*8)-base) for j in range(12)]})
if components:ammo.append({'index':index,'components':components})
except (OSError,ValueError):continue
report['originalAmmoModules']=ammo
args.output.parent.mkdir(parents=True, exist_ok=True)
with args.output.open('x', encoding='utf-8') as output:
json.dump(report, output, indent=2)
print(json.dumps({'player': components[0]['entityId'], 'components': [row['name'] for row in components], 'physicsBindings': len(physics),
'poseLookup': {k: v for k, v in pose_lookup.items() if k not in ('metadataHeader', 'rootHeader')}, 'gameInputsSent': 0, 'screenUse': False}))
finally:
kernel.CloseHandle(process)
if __name__ == '__main__':
main()
"""Bounded external reads of the current player's build-pinned flight components.
No injection, hooks, native calls, process writes, input or screen access.
The player's identity is obtained from the explicitly started toolbox observer.
"""
import argparse
import ctypes as c
import ctypes.wintypes as w
import hashlib
import json
import math
from pathlib import Path
import struct
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--pid', type=int, required=True)
parser.add_argument('--profile', type=Path, required=True)
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--export-density',type=Path,help='Export only the original immutable Galactic density resource, using VM_READ')
parser.add_argument('--bootstrap-components', action='store_true', help='Read unique type-8 player candidates while diagnosing unavailable pose telemetry; never accepts flight input')
parser.add_argument('--damage',action='store_true',help='Include the build-pinned BasicDamage component using bounded read-only access')
args = parser.parse_args()
profile = json.loads(args.profile.read_text(encoding='utf-8-sig'))
image = Path(profile['image']).resolve(strict=True)
if hashlib.sha256(image.read_bytes()).hexdigest() != profile['sha256']:
raise ValueError('Game build mismatch')
kernel = c.WinDLL('kernel32', use_last_error=True)
kernel.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]
kernel.OpenProcess.restype = w.HANDLE
kernel.CloseHandle.argtypes = [w.HANDLE]
kernel.QueryFullProcessImageNameW.argtypes = [w.HANDLE, w.DWORD, w.LPWSTR, c.POINTER(w.DWORD)]
kernel.ReadProcessMemory.argtypes = [w.HANDLE, c.c_void_p, c.c_void_p, c.c_size_t, c.POINTER(c.c_size_t)]
kernel.CreateToolhelp32Snapshot.argtypes = [w.DWORD, w.DWORD]
kernel.CreateToolhelp32Snapshot.restype = w.HANDLE
class Module(c.Structure):
_fields_ = [('dwSize', w.DWORD), ('th32ModuleID', w.DWORD), ('th32ProcessID', w.DWORD),
('GlblcntUsage', w.DWORD), ('ProccntUsage', w.DWORD), ('modBaseAddr', c.c_void_p),
('modBaseSize', w.DWORD), ('hModule', w.HMODULE), ('szModule', w.WCHAR * 256), ('szExePath', w.WCHAR * 260)]
kernel.Module32FirstW.argtypes = [w.HANDLE, c.POINTER(Module)]
kernel.Module32NextW.argtypes = [w.HANDLE, c.POINTER(Module)]
kernel.WaitNamedPipeW.argtypes = [w.LPCWSTR, w.DWORD]
kernel.CreateFileW.argtypes = [w.LPCWSTR, w.DWORD, w.DWORD, c.c_void_p, w.DWORD, w.DWORD, w.HANDLE]
kernel.CreateFileW.restype = w.HANDLE
kernel.WriteFile.argtypes = [w.HANDLE, c.c_void_p, w.DWORD, c.POINTER(w.DWORD), c.c_void_p]
kernel.ReadFile.argtypes = [w.HANDLE, c.c_void_p, w.DWORD, c.POINTER(w.DWORD), c.c_void_p]
process = kernel.OpenProcess(0x410, False, args.pid)
if not process:
raise c.WinError(c.get_last_error())
def read(at, size):
if at < 0x10000 or not 0 < size <= 8192:
raise ValueError('Read outside bounded component memory')
value = c.create_string_buffer(size)
done = c.c_size_t()
if not kernel.ReadProcessMemory(process, at, value, size, c.byref(done)) or done.value != size:
raise c.WinError(c.get_last_error())
return value.raw
def number(at, kind='<Q'):
return struct.unpack(kind, read(at, struct.calcsize(kind)))[0]
def pipe_call(op):
name = rf'\\.\pipe\XFEEDToolBox.Navigation.{args.pid}'
if not kernel.WaitNamedPipeW(name, 2000):
raise c.WinError(c.get_last_error())
pipe = kernel.CreateFileW(name, 0xC0000000, 0, None, 3, 0, None)
if pipe == c.c_void_p(-1).value:
raise c.WinError(c.get_last_error())
try:
count = w.DWORD()
body = (json.dumps({'op': op}) + '\n').encode()
if not kernel.WriteFile(pipe, body, len(body), c.byref(count), None):
raise c.WinError(c.get_last_error())
data = c.create_string_buffer(65536)
if not kernel.ReadFile(pipe, data, len(data), c.byref(count), None):
raise c.WinError(c.get_last_error())
reply = json.loads(data.raw[:count.value])
if not reply.get('ok'):
raise ValueError(reply.get('error', 'Observer operation failed'))
return reply['result']
finally:
kernel.CloseHandle(pipe)
try:
path = c.create_unicode_buffer(32768)
length = w.DWORD(len(path))
if not kernel.QueryFullProcessImageNameW(process, 0, path, c.byref(length)) or Path(path.value).resolve() != image:
raise ValueError('Wrong selected process image')
modules = kernel.CreateToolhelp32Snapshot(0x18, args.pid)
if modules == c.c_void_p(-1).value:
raise c.WinError(c.get_last_error())
base = image_size = 0
try:
item = Module()
item.dwSize = c.sizeof(Module)
found = kernel.Module32FirstW(modules, c.byref(item))
while found:
if Path(item.szExePath).resolve() == image:
base, image_size = item.modBaseAddr, item.modBaseSize
break
found = kernel.Module32NextW(modules, c.byref(item))
finally:
kernel.CloseHandle(modules)
if not base:
raise ValueError('Game image was not found in selected process')
description = pipe_call('describe')
if description['processId'] != args.pid or description['gameSha256'] != profile['sha256']:
raise ValueError('Observer identity mismatch')
player = {'inGame': description['inGame'], 'entityId': None, 'poseVerified': False} if args.bootstrap_components else pipe_call('read')
if not player['inGame']:
raise ValueError('Player is outside the game')
entity = int(player['entityId']) if player['entityId'] is not None else None
def in_image(at):
return base <= at < base + image_size
def lifecycle(obj):
owner = number(obj + 0x20)
method = number(number(owner) + 0x20)
if not in_image(method):
return None
code = read(method, 12)
reference = 0
if code[:3] == b'\x48\x8d\x81' and code[7] == 0xC3:
reference = owner + struct.unpack_from('<i', code, 3)[0]
elif code[:3] == b'\x48\x8d\x41' and code[4] == 0xC3:
reference = owner + struct.unpack_from('<b', code, 3)[0]
elif code[:3] == b'\x48\x8b\x81' and code[7] == 0xC3:
reference = number(owner + struct.unpack_from('<i', code, 3)[0])
elif code[:3] == b'\x48\x8b\x41' and code[4] == 0xC3:
reference = number(owner + struct.unpack_from('<b', code, 3)[0])
if not reference:
return None
life = number(reference)
return life if number(life + 0xC0, '<I') == 8 and (entity is None or number(life + 0xB0) == entity) else None
components = []
pools=[('heat', profile['heatPool'], 0x190), ('heatManager', profile['heatManagerPool'], 0x228), ('fsd', profile['fsdPool'], 0xEC0)]
if args.damage:
if profile['sha256']!='6891f70a021fa9f3a0e00458c284f48d1226e68ae7fc02ad48081aed47857e93':raise ValueError('Damage layout not known for this build')
pools.append(('damage',0x5FCF500,0x3E8))
pools.append(('docking',0x5FAA2A0,0x750))
for label, rva, size in pools:
pool = base + rva
if number(pool + 0x68) != size or number(pool + 0x90) > 128:
raise ValueError('Component pool size/bound mismatch')
block, visited = number(pool + 0x30), set()
while block and block not in visited and len(visited) < 128:
visited.add(block)
count = number(block + 0x38)
if not 1 <= count <= 128:
raise ValueError('Component block count invalid')
for index in range(count):
obj = block + 0x80 + index * size
try:
if not in_image(number(obj)):
continue
life = lifecycle(obj)
if life is None:
continue
tables = []
for offset in (0, 8, 0x58, 0x60):
table = number(obj + offset)
if not in_image(table):
continue
methods = [number(table + i * 8) for i in range(80)]
tables.append({'objectOffset': hex(offset), 'rva': hex(table - base), 'methods': [hex(m - base) if in_image(m) else None for m in methods]})
components.append({'name': label, 'address': hex(obj), 'lifecycle': hex(life), 'entityId': str(number(life + 0xB0)), 'ownerContext': hex(number(obj + 0x10)), 'tables': tables, 'bytes': read(obj, size).hex()})
except (OSError, ValueError):
continue
block = number(block + 0x20)
if args.bootstrap_components:
candidates=[]
for row in components:
if row['name']!='fsd':continue
try:
fsd_obj=int(row['address'],16);physics_root=number(fsd_obj+0x2A8)-0x58
input_root=number(physics_root+0x3B0)-0x60;controller=number(input_root+0x2F8)
if not in_image(number(controller)) or number(number(controller)+0x158)!=base+0x25E5DD0:continue
local_player=number(input_root+0x2E8);local_table=number(local_player);controlled_getter=number(local_table+0x160)
local_source={'address':hex(local_player),'tableRva':hex(local_table-base),'getControlledRva':hex(controlled_getter-base),
'getterBytes':read(controlled_getter,32).hex(),'header':read(local_player,0x800).hex()}
if read(controlled_getter,8)==bytes.fromhex('488d8138010000c3'):
pilot_life=number(local_player+0x138);pilot_root=number(pilot_life+0xC8)-0x30
pilot_type=number(base+0x5F05BA4,'<I');pilot_meta=number(pilot_root+0x10)
pilot_slot=number(number(pilot_meta+0x30)+pilot_type*16)
if pilot_slot<number(pilot_meta+0x18):
pilot_interface=number(number(pilot_root+0x18)+pilot_slot*8)
ship_getter=number(number(pilot_interface)+0x68)
local_source['pilot']={'life':hex(pilot_life),'root':hex(pilot_root),'typeId':pilot_type,'denseSlot':pilot_slot,
'interface':hex(pilot_interface),'tableRva':hex(number(pilot_interface)-base),'getShipRva':hex(ship_getter-base),
'getterBytes':read(ship_getter,256).hex(),'header':read(pilot_interface,0x400).hex()}
actor_owner=pilot_interface-0x58;actor_count=number(actor_owner+0x970);actor_array=number(actor_owner+0x978)
if not 0<actor_count<=32:raise ValueError('Local player actor list exceeds bound')
actors=[]
for actor_index in range(actor_count):
actor=actor_array+actor_index*0x1C0;actor_life=number(actor+0x1B0)
actors.append({'kind':number(actor+0x190,'<B'),'ticket':number(actor+0x1B8,'<i'),
'life':hex(actor_life),'entityId':str(number(actor_life+0xB0)) if actor_life else None,
'lifeType':number(actor_life+0xC0,'<I') if actor_life else None})
eligible=[row for row in actors if row['ticket']!=-1]
chosen=next((row for kind in (2,0,3) for row in eligible if row['kind']==kind),None)
local_source['pilot']['actors']=actors
local_source['pilot']['selectedPlayerShip']=chosen
local_source['pilot']['matchesCandidate']=chosen is not None and chosen['life']==row['lifecycle']
candidates.append({'entityId':row['entityId'],'lifecycle':row['lifecycle'],'fsd':row['address'],
'physics':hex(physics_root),'inputRoot':hex(input_root),'controller':hex(controller),'controllerTableRva':hex(number(controller)-base),
'controllerHeader':read(controller,0xC00).hex(),'inputHeader':read(input_root,0x400).hex(),
'physicsHeader':read(physics_root,0xB00).hex(),'localPlayerSource':local_source})
except (OSError,ValueError):continue
report={'pid':args.pid,'base':hex(base),'gameSha256':profile['sha256'],'unclassifiedType8Candidates':candidates,
'components':components,'gameInputsSent':0,'screenUse':False,'processAccess':'QUERY_INFORMATION | VM_READ'}
with args.output.open('x',encoding='utf-8') as output:json.dump(report,output,indent=2)
print(json.dumps({'unclassifiedCandidates':[{k:v for k,v in row.items() if not k.endswith('Header')} for row in candidates]}))
return
primary_components=[row for row in components if row['name'] in ('heat','heatManager','fsd')]
if len(primary_components) != 3 or len({row['ownerContext'] for row in components}) != 1 or len({row['entityId'] for row in components}) != 1:
raise ValueError('The three unique player components did not match')
life = int(components[0]['lifecycle'], 16)
root = number(life + 0xC8) - 0x30
getter = base + 0x8AE6F0
code = read(getter, 12)
if code[:2] != b'\x8b\x05' or code[6:12] != b'\x89\x02\x48\x8b\xc2\xc3':
raise ValueError('Pose ID getter signature mismatch')
type_id = number(getter + 6 + struct.unpack_from('<i', code, 2)[0], '<I')
if not 0 < type_id < 100000:
raise ValueError('Pose type ID outside diagnostic bound')
metadata = number(root + 0x10)
slots = number(metadata + 0x30)
slot = number(slots + type_id * 16)
if not 0 <= slot < 512:
raise ValueError('Resolved pose slot outside diagnostic bound')
pose = number(number(root + 0x18) + slot * 8)
pose_getter = number(number(pose) + 0x28)
pose_lookup = {'root': hex(root), 'metadata': hex(metadata), 'poseTypeId': type_id, 'denseSlotIndex': slot,
'poseAddress': hex(pose), 'getterRva': hex(pose_getter - base), 'getterMatches': pose_getter == base + 0x3CBC080,
'metadataHeader': read(metadata, 128).hex(), 'rootHeader': read(root, 128).hex()}
physics = []
component_count = number(metadata + 0x18)
if not 1 <= component_count <= 1024:
raise ValueError('Dense component count outside diagnostic bound')
array = number(root + 0x18)
if args.damage:
interfaces=[]
for dense_index in range(component_count):
try:
candidate=number(array+dense_index*8);table=number(candidate)
if not in_image(table):continue
methods=[number(table+i*8) for i in range(80)]
interfaces.append({'index':dense_index,'address':hex(candidate),'tableRva':hex(table-base),'header':read(candidate,128).hex(),
'methods':[hex(m-base) if in_image(m) else None for m in methods]})
except (OSError,ValueError):continue
player['readOnlyInterfaces']=interfaces
damage_type=number(base+0x5F2F254,'<I')
damage_slot=number(number(metadata+0x30)+damage_type*16)
if damage_slot<component_count:
damage=number(array+damage_slot*8);encoded=number(damage+0x178)
property_object=number(damage-0x58+0x120);property_table=number(property_object)
import math
value=struct.unpack('<f',struct.pack('<I',(encoded>>32)^(encoded&0xffffffff)))[0]
player['readOnlyDamage']={'interface':hex(damage),'tableRva':hex(number(damage)-base),'typeId':damage_type,'denseSlot':damage_slot,
'numerator':value if math.isfinite(value) else None,'property':hex(property_object),'propertyTableRva':hex(property_table-base),
'propertyGetterRva':hex(number(property_table+0x18)-base),'propertyHeader':read(property_object,128).hex(),
'header':read(damage-0x58,0x3e8).hex()}
inspected = set()
for index in range(component_count):
candidate = number(array + index * 8)
for adjustment in (0, 8, 0x58, 0x60):
obj = candidate - adjustment
if obj in inspected:
continue
inspected.add(obj)
try:
if not in_image(number(obj)) or lifecycle(obj) != life:
continue
input_root = number(obj + 0x3B0) - 0x60
if not in_image(number(input_root)):
continue
controller = number(input_root + 0x2F8)
if number(number(controller) + 0x158) != base + 0x25E5DD0:
continue
physics.append({'address': hex(obj), 'denseSlot': index, 'inputRoot': hex(input_root),
'inputTableRva': hex(number(input_root) - base), 'controller': hex(controller),
'controllerTableRva': hex(number(controller) - base), 'inputBytes': read(input_root, 0x580).hex(),
'physicsBytes': read(obj, 0xB00).hex(), 'controllerMethods': [hex(number(number(controller) + i * 8) - base) for i in range(64)]})
except (OSError, ValueError):
continue
report = {'pid': args.pid, 'gameSha256': profile['sha256'], 'base': hex(base), 'bootstrapCandidatesOnly': args.bootstrap_components,
'player': player, 'poseLookup': pose_lookup, 'components': components, 'physics': physics,
'poseBytes': read(pose, 0x480).hex(), 'gameInputsSent': 0, 'screenUse': False, 'processAccess': 'QUERY_INFORMATION | VM_READ'}
maintenance_pool=base+0x5F85A60
maintenance=dict(stride=number(maintenance_pool+0x68),count=number(maintenance_pool+0x90),objects=[])
if maintenance['stride']==0x440 and maintenance['count']<=128:
block=number(maintenance_pool+0x30);visited=set()
while block and block not in visited and len(visited)<128:
visited.add(block);count=number(block+0x38)
if not 1<=count<=128:raise ValueError('Maintenance pool block exceeds bounds')
for index in range(count):
obj=block+0x80+index*0x440
if number(obj) not in (base+0x52409D0,base+0x5240888):continue
table=number(obj+0x58)
maintenance['objects'].append(dict(address=hex(obj),interface=hex(obj+0x58),table=hex(table-base),header=read(obj,0x440).hex(),
methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(40)]]))
block=number(block+0x20)
report['maintenancePool']=maintenance
fsd_address = int(next(row for row in components if row['name'] == 'fsd')['address'], 16)
sources = []
for label, offset in [('worldTransform', 0x320), ('navigationTarget', 0x338), ('spaceBodies', 0x6A0), ('actualFlightControl',0x2A8), ('heatInput',0x2F0),
('localFlightState',0x3F8),('cruiseTargetService',0x470),('playerController',0x6B0),('fsdEligibility',0x6D0)]:
interface = number(fsd_address + offset)
table = number(interface)
if not in_image(table):
raise ValueError('FSD source interface table outside selected image')
methods = [number(table + i*8) for i in range(112)]
sources.append({'name': label, 'address': hex(interface), 'tableRva': hex(table-base),
'methods': [hex(m-base) if in_image(m) else None for m in methods], 'header': read(interface, 512).hex()})
report['fsdSources'] = sources
# Original galaxy route interfaces, looked up through registered component
# IDs. This records pointers and method tables only; no native invocation.
route_roots = [('ship', root)]
if len(physics) == 1:
input_root = int(physics[0]['inputRoot'], 16)
local_player = number(input_root + 0x2E8)
controlled_getter = number(number(local_player) + 0x160)
if read(controlled_getter, 8) == bytes.fromhex('488d8138010000c3'):
pilot_life = number(local_player + 0x138)
route_roots.append(('pilot', number(pilot_life + 0xC8) - 0x30))
route_bindings = []
route_components = []
for root_name, route_root in route_roots:
route_meta = number(route_root + 0x10)
route_slots = number(route_meta + 0x30)
route_array = number(route_root + 0x18)
route_count = number(route_meta + 0x18)
if not 1 <= route_count <= 1024:
raise ValueError('Route component registry exceeds its bound')
for index in range(route_count):
component = number(route_array + index * 8)
try:
table = number(component)
if in_image(table):
item=dict(owner=root_name, denseSlot=index,
address=hex(component), tableRva=hex(table-base), header=read(component, 0x80).hex())
if args.damage:
item['methods']=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(80)]]
header=read(component-0x58,0x458)
for offset in range(0,len(header)-7,8):
if struct.unpack_from('<Q',header,offset)[0]==base+0x5322218:
ui=component-0x58+offset;assist=number(ui+0x258);at=number(assist)
if in_image(at):
item['supercruiseUi']=dict(address=hex(ui),assist=hex(assist),assistTable=hex(at-base),
assistHeader=read(assist,0x600).hex(),assistMethods=[hex(v-base) if in_image(v) else None for v in [number(at+i*8) for i in range(32)]])
route_components.append(item)
except (OSError, ValueError):
continue
for name, id_rva in [('route',0x5F05DD0), ('routeController',0x5F59D90),
('mapRouteRequest',0x5F05DD4), ('mapRoutePlotter',0x5F05DCC),
('world',0x5F05D3C), ('targets',0x5F44D04), ('galaxy',0x5F05CC8),
('galaxyCatalogue',0x5F05C1C),('jumpConfiguration',0x5F05CB4),('supercruiseAssist',0x5F53170),
('cruiseControlGeometry',0x5F53174),('cruiseArrival',0x5F34640),('cruiseNavigationWorld',0x5F34660),('cruiseInputTransform',0x5F2F27C),
('flightModeInterface',0x5F2F260),('dockingMotionInterface',0x5F44D48),('dockingGuideInterface',0x5F44CF0),
('dockingAssist',0x5F53178),('launchAssist',0x5F5317C),('shipActivityController',0x5F44C98),
('cargoInventory',0x5F2F25C),('fuelInventory',0x5F45128),('maintenanceOperations',0x5F7E800),
('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC)]:
type_id = number(base + id_rva, '<I')
if not 0 < type_id < 100000:
continue
index = number(route_slots + type_id * 16)
if index >= route_count:
continue
interface = number(route_array + index * 8)
if interface < 0x10000:
continue
table = number(interface)
if not in_image(table):
continue
binding_row=dict(owner=root_name, name=name, root=hex(route_root),
typeId=type_id, denseSlot=index, address=hex(interface), tableRva=hex(table-base),
methods=[hex(method-base) if in_image(method) else None
for method in [number(table+i*8) for i in range(64)]], header=read(interface, 0x700).hex())
if name in ('supercruiseAssist','cruiseControlGeometry'):
parameters=number(interface-8)
if parameters:binding_row['parameters']=dict(address=hex(parameters),data=read(parameters,0x100).hex())
if name=='launchAssist':
bindings=number(interface+0x230)
if bindings:
table=number(bindings)
binding_row['inputBindings']=dict(address=hex(bindings),table=hex(table-base),
getBinding=hex(number(table+0x40)-base),header=read(bindings,0x80).hex())
brain=number(interface+0x1D0)
if brain and in_image(number(brain)):
getter=number(number(brain)+0xA0)
if in_image(getter):binding_row['brain']=dict(address=hex(brain),table=hex(number(brain)-base),
dutyGetter=hex(getter-base),dutyGetterCode=read(getter,64).hex(),header=read(brain,0x200).hex())
if read(getter,8)==bytes.fromhex('488b81e8010000c3'):
duty=number(brain+0x1E8)
if duty and in_image(number(duty)):
table=number(duty)
binding_row['brain']['duty']=dict(address=hex(duty),table=hex(table-base),typeGetter=hex(number(table+0x20)-base))
if table==base+0x5156130:
binding_row['brain']['duty']['header']=read(duty,0x378).hex()
refs=[]
for label,offset in [('station',0x300),('pad',0x318),('approach',0x330)]:
cached=number(duty+offset+0x10)
if cached>0x10000 and in_image(number(cached)):
cached_table=number(cached)
refs.append(dict(name=label,address=hex(cached),table=hex(cached_table-base),
header=read(cached,0x200).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(cached_table+i*8) for i in range(48)]]))
binding_row['brain']['duty']['typedComponents']=refs
control=number(interface+0x240)
if control and in_image(number(control)):
table=number(control)
binding_row['desiredInput']=dict(address=hex(control),table=hex(table-base),
setter=hex(number(table+0x158)-base),clearer=hex(number(table+0x150)-base),throttle=number(control+0x680,'<f'))
if name=='supercruiseAssist':
main_table=number(interface-0x58)
if in_image(main_table):binding_row['mainTable']=hex(main_table-base);binding_row['mainMethods']=[hex(v-base) if in_image(v) else None for v in [number(main_table+i*8) for i in range(32)]]
internals=[]
for label,offset in [('approachGeometry',0x90),('flightController',0xF0),('inputs',0x198),('throttleBinding',0x258)]:
source=number(interface-0x58+offset)
if not source:continue
vt=number(source)
internals.append(dict(name=label,address=hex(source),table=hex(vt-base) if in_image(vt) else None,header=read(source,0x300).hex(),
methods=[hex(v-base) if in_image(v) else None for v in [number(vt+i*8) for i in range(64)]] if in_image(vt) else []))
binding_row['internals']=internals
if name=='jumpConfiguration' and number(table+0x130)==base+0x2C9A7A0:
holder=number(interface+0x78)
if holder and number(holder+0x10):
owner_data=number(holder+0x18);configuration=number(owner_data+0x270)
if configuration:binding_row['jumpParameters']=dict(address=hex(configuration),data=read(configuration,64).hex(),floats=list(struct.unpack('<14f',read(configuration,56))))
route_bindings.append(binding_row)
report['routeBindings'] = route_bindings
global_lookup=[]
for binding in route_bindings:
if binding['owner']!='pilot' or binding['name']!='route':continue
source=int(binding['address'],16)-0x58+8
for step,slot in enumerate([0,0x20]):
method=number(number(source)+slot); code=read(method,16)
row=dict(step=step,address=hex(source),method=hex(method-base),code=code.hex())
if code[:3]==b'\x48\x8b\x41' and code[4]==0xc3:source=number(source+struct.unpack_from('<b',code,3)[0])
elif code[:3]==b'\x48\x8b\x81' and code[7]==0xc3:source=number(source+struct.unpack_from('<i',code,3)[0])
elif code[:3]==b'\x48\x8d\x41' and code[4]==0xc3:source+=struct.unpack_from('<b',code,3)[0]
elif code[:3]==b'\x48\x8d\x81' and code[7]==0xc3:source+=struct.unpack_from('<i',code,3)[0]
else:global_lookup.append(row);break
row['result']=hex(source);global_lookup.append(row)
if source>0x10000 and in_image(number(source)):
global_lookup.append(dict(source=hex(source),contextId=number(base+0x5F05AB8,'<I'),
table=hex(number(source)-base),methods=[hex(number(number(source)+i*8)-base) for i in range(8)],header=read(source,128).hex()))
count=number(source+0x28); buckets=number(source+0x20);context_id=number(base+0x5F05AB8,'<I')
if 0<count<=1024 and number(number(source))==base+0x529a30:
for bucket in range(count):
sentinel=buckets+bucket*8; node=number(sentinel); seen=set()
while node and node!=sentinel and node not in seen and len(seen)<1024:
seen.add(node)
for label,id_rva in [('stationMaintenance',0x5F05ACC),('stationServices',0x5F05B94),('economyCatalog',0x5F05CD0),('commanderServices',0x5F05BAC),('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC)]:
if number(node+8,'<I')==number(base+id_rva,'<I'):
service=number(node+16);table=number(service)
if in_image(table):global_lookup.append(dict(domainService=label,typeId=number(node+8,'<I'),address=hex(service),table=hex(table-base),header=read(service,0x600).hex(),methods=[hex(number(table+i*8)-base) if in_image(number(table+i*8)) else None for i in range(80)]))
if number(node+8,'<I')==context_id:
ctx=number(node+16); method=number(number(ctx)+0x18)
code=read(method,16)
row=dict(context=hex(ctx),method=hex(method-base),code=code.hex(),header=read(ctx,128).hex())
if code[:3]==b'\x48\x8b\x41' and code[4]==0xc3:row['value']=hex(number(ctx+struct.unpack_from('<b',code,3)[0]))
elif code[:3]==b'\x48\x8b\x81' and code[7]==0xc3:row['value']=hex(number(ctx+struct.unpack_from('<i',code,3)[0]))
global_lookup.append(row)
if method==base+0x1f3150:
weak=number(ctx+0x568); entity_component=number(weak+0xC8) if weak else 0
if entity_component:
global_root=entity_component-0x30; meta=number(global_root+0x10);slots=number(meta+0x30);array=number(global_root+0x18);n=number(meta+0x18)
if not 1<=n<=2048:raise ValueError('Global component registry outside bound')
row['globalRoot']=hex(global_root); row['globalCount']=n
display_matches=[]
for component_index in range(n):
candidate=number(array+component_index*8)
if candidate<0x10000:continue
try:
header=read(candidate,2048);at=header.find(struct.pack('<I',0xd72f6ba2))
if at>=0:display_matches.append(dict(slot=component_index,address=hex(candidate),table=hex(number(candidate)-base),gdmOffset=hex(at),header=header.hex()))
except (OSError,ValueError):continue
row['galaxyDisplayComponents']=display_matches
global_catalogue=[]
for label,rva in [('galaxy',0x5F05C1C),('galaxyNames',0x5F05C20),('catalogue',0x5F05C24),('galaxyService',0x5F05CC8),('stationMaintenance',0x5F05ACC),('stationServices',0x5F05B94),('economyCatalog',0x5F05CD0),('commanderServices',0x5F05BAC),('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC),('commodityDefinitions',0x5F05C34)]:
kind=number(base+rva,'<I');index=number(slots+kind*16)
if index<n:
obj=number(array+index*8)
if obj<0x10000:continue
table=number(obj)
if not in_image(table):continue
entry=dict(name=label,typeId=kind,address=hex(obj),table=hex(table-base),header=read(obj,1024).hex(),methods=[hex(number(table+i*8)-base) for i in range(80)])
if label=='galaxy':
origin=number(obj+0x188)
if origin>0x10000:entry['origin']=list(struct.unpack('<iii',read(origin,12)))
density=number(obj+0x100)
if density>0x10000:
grid=number(density+0x78)
entry['density']=dict(address=hex(density),header=read(density,144).hex(),grid=hex(grid))
entry['density']['mips']=[]
for mip in range(8):
raster=number(density+0x38+mip*8)
if raster>0x10000:entry['density']['mips'].append(dict(level=mip,address=hex(raster),header=read(raster,32).hex(),width=number(raster,'<I'),height=number(raster+4,'<I')))
if grid>0x10000:entry['density'].update(gridHeader=read(grid,32).hex(),width=number(grid,'<I'),height=number(grid+4,'<I'))
if args.export_density and grid>0x10000:
grid=number(density+0x38)
width=number(grid,'<I');height=number(grid+4,'<I');rows=number(grid+0x10)
if not 16<=width<=8192 or not 16<=height<=8192 or width*height>16777216:raise ValueError('Galactic raster exceeds bound')
pixels=bytearray(width*height*4)
for y in range(height):
density_row=number(rows+y*8);x=0;blocks=0
while x<width and blocks<=width:
blocks+=1;header=number(density_row,'<I');density_row+=4;length=header&0x7fffffff
if not 0<length<=width-x:raise ValueError('Invalid Galactic density RLE')
if header&0x80000000:
value=read(density_row,4);density_row+=4;chunk=value*length
else:
chunk=read(density_row,length*4);density_row+=length*4
start=(y*width+x)*4;pixels[start:start+length*4]=chunk;x+=length
if x!=width:raise ValueError('Incomplete Galactic density row')
sx=number(density+4,'<I');sz=number(density+0x10,'<I')
geometry=(entry['origin'][0]/32,entry['origin'][2]/32,width*2147483648*10/sx,height*2147483648*10/sz)
args.export_density.parent.mkdir(parents=True,exist_ok=True)
import array as density_array
original=memoryview(pixels).cast('I');maximum=max(original)
normalized=density_array.array('f',(value/maximum for value in original)).tobytes()
args.export_density.write_bytes(b'XFEGDM2\0'+struct.pack('<II4d',width,height,*geometry)+normalized)
entry['density']['exported']=dict(file=str(args.export_density),dimensions=[width,height],geometry=geometry,maximum=maximum,sha256=hashlib.sha256(pixels).hexdigest())
cached=[]
for wanted in [int(player['geometry']['worldAddress'])&0x7fffffffffffff,182359951707,10477373803]:
node=number(obj+0x370);depth=0
while node and depth<64:
depth+=1;key=number(node+0x20)
if key==wanted:
galaxy_object=number(node+0x28)
if galaxy_object:
compact=number(galaxy_object+0x38)
if compact:cached.append(dict(system=str(wanted),object=hex(galaxy_object),compact=hex(compact),data=read(compact,104).hex()))
break
node=number(node+(8 if key<wanted else 0))
entry['cachedRecords']=cached
engine=number(obj+0x228)
if engine>0x10000:
vt=number(engine)
if in_image(vt):entry['generator']=dict(address=hex(engine),table=hex(vt-base),header=read(engine,512).hex(),methods=[hex(number(vt+i*8)-base) for i in range(24)])
global_catalogue.append(entry)
row['catalogues']=global_catalogue
node=number(node)
report['globalLookup']=global_lookup
report['registeredRouteComponents'] = route_components
route_lists=[]
for binding in route_bindings:
if binding['owner']!='pilot' or binding['name']!='route':
continue
sentinel=int(binding['address'],16)+0x110
directions=[]
for direction in (0,8):
node=number(sentinel+direction);addresses=[];seen=set()
while node!=sentinel and node and len(seen)<25000:
if node in seen:raise ValueError('Unexpected cycle in game route list')
seen.add(node);addresses.append(str(number(node+0x10)));node=number(node+direction)
directions.append(dict(direction=direction,terminated=node==sentinel,addresses=addresses))
route_lists.append(dict(sentinel=hex(sentinel),directions=directions))
report['originalRouteLists']=route_lists
# D build's FuelScoopModuleComponent registration: native size 0x210 and pool RVA.
# Keep optional diagnostics separate from the three required ship components.
extra=[];unassociated=[]
scoop_pool=base+0x5FA81A0
report['scoopPoolMetadata']={'rva':'0x5fa81a0','stride':number(scoop_pool+0x68),'count':number(scoop_pool+0x90),'header':read(scoop_pool,0xC0).hex()}
if number(scoop_pool+0x68)==0x210 and number(scoop_pool+0x90)<=128:
block=number(scoop_pool+0x30);visited=set()
while block and block not in visited and len(visited)<128:
visited.add(block);count=number(block+0x38)
if not 1<=count<=128:raise ValueError('Fuel scoop block count exceeds bound')
for index in range(count):
obj=block+0x80+index*0x210
table=number(obj)
if not in_image(table):continue
methods=[number(table+i*8) for i in range(80)]
row={'name':'fuelScoop','address':hex(obj),'ownerContext':hex(number(obj+0x10)),
'tableRva':hex(table-base),'methods':[hex(m-base) if in_image(m) else None for m in methods],'header':read(obj,0x210).hex()}
if number(obj+0x10)==int(components[0]['ownerContext'],16):extra.append(row)
elif len(unassociated)<16:
owner=number(obj+0x20)
row['ownerHeader']=read(owner,0x100).hex()
row['ownerAddress']=hex(owner)
row['matchingPlayerLifecycle']=lifecycle(obj) is not None
unassociated.append(row)
block=number(block+0x20)
report['extraComponents']=extra
report['unassociatedFuelScoops']=unassociated
manager = int(next(row for row in components if row['name'] == 'heatManager')['address'], 16)
modifier_count, modifier_array = number(manager + 0x160), number(manager + 0x168)
if not 0 <= modifier_count <= 32 or modifier_count and not modifier_array:
raise ValueError('Heat modifier list exceeds the bounded observation range')
modifiers = []
for index in range(modifier_count):
modifier = number(modifier_array + index*8)
table = number(modifier)
if not in_image(table):
raise ValueError('Heat modifier table outside the selected game image')
methods = [number(table + i*8) for i in range(12)]
modifiers.append({'index':index, 'address':hex(modifier), 'tableRva':hex(table-base),
'methods':[hex(method-base) if in_image(method) else None for method in methods],
'header':read(modifier, 512).hex()})
if number(manager + 0x160) != modifier_count or number(manager + 0x168) != modifier_array:
raise ValueError('Heat modifier list changed while reading')
report['heatModifiers'] = modifiers
navigation = int(next(row for row in sources if row['name'] == 'navigationTarget')['address'], 16)
nav_count, nav_capacity, nav_storage = number(navigation + 0x2C0), number(navigation + 0x2C8), number(navigation + 0x2D0)
if nav_count > 1024 or nav_capacity > 16384 or nav_count > nav_capacity or nav_count and not nav_storage:
raise ValueError('Navigation record collection exceeds the bounded observation range')
records = []
if nav_count:
if nav_capacity == 0 or nav_capacity & (nav_capacity-1):
raise ValueError('Navigation record hash capacity is not a power of two')
occupied = read(nav_storage, (nav_capacity+7)//8)
entries = nav_storage + (((nav_capacity>>3)+7)&~7)
def name_at(record, offset):
try:
text = number(record + offset)
if not text: return ''
length = number(text)
if not 0 < length <= 128: return ''
return read(text+0x14, length).decode('utf-8').replace('\r',' ').replace('\n',' ')
except (OSError, ValueError, UnicodeError): return ''
for index in range(nav_capacity):
if not occupied[index>>3] & (1<<(index&7)): continue
entry=entries+index*16
record=number(entry+8)
if not record: continue
records.append({'key':number(entry,'<i'),'address':hex(record),
'worldAddress':str(number(record+0x20)),
'name':name_at(record,0x140),'secondaryName':name_at(record,0x148),
'kind':number(record+0x184,'<I'),'type':number(record+0x188,'<I'),
'header':read(record,0x1E0).hex()})
if len(records) != nav_count:
raise ValueError('Navigation records changed or had null entries during observation')
if number(navigation+0x2C0)!=nav_count or number(navigation+0x2C8)!=nav_capacity or number(navigation+0x2D0)!=nav_storage:
raise ValueError('Navigation records changed during observation')
report['navigationRecords']={'count':nav_count,'selectedKey':number(navigation+0x38C,'<i'),'records':records}
if args.damage:
# The original life-reference lookup (0x62D120) uses this world's
# entity directory. Check the own-ship key/value before trusting it.
player_life=int(components[0]['lifecycle'],16)
context=number(player_life+0x38)
directory=dict(context=hex(context),header=read(context+0x140,0x30).hex())
count,capacity,storage=number(context+0x148),number(context+0x150),number(context+0x158)
directory.update(count=count,capacity=capacity,storage=hex(storage))
if count<=capacity and 8<=capacity<=65536 and not capacity&(capacity-1) and storage:
occupied=read(storage,(capacity+7)//8);entries=storage+(((capacity>>3)+7)&~7)
markets={struct.unpack_from('<I',bytes.fromhex(r['header']),0x130)[0] for r in records if r['kind'] in (6,12,14,18,28,35)}
matches=[];own=False
for i in range(capacity):
if not occupied[i>>3]&(1<<(i&7)):continue
at=entries+i*16;key,value=number(at),number(at+8)
if key==number(player_life+0xB0):own=value==player_life
if key in markets:matches.append(dict(key=str(key),life=hex(value),type=number(value+0xC0,'<I'),entityId=str(number(value+0xB0))))
directory.update(ownShipMatches=own,stationMatches=matches)
report['entityDirectory']=directory
# Independently reproduce only the immutable affine transforms used
# by the ATC distance guard; no original getter or request is invoked.
def registered(root,type_id):
meta=number(root+0x10);count=number(meta+0x18);slots=number(meta+0x30)
slot=number(slots+type_id*16)
return number(number(root+0x18)+slot*8) if 0<count<=1024 and slot<count else 0
def body_hash(key):
mask=(1<<64)-1;value=(key*0x40000+(~key&mask))&mask
value=((value>>31^value)*21)&mask;value=((value>>11^value)*65)&mask
return ((value>>22)^value)&0xffffffff
def body_entry(table,capacity,key,payload,system=False):
if not table or not 0<capacity<=65536:return 0
sentinel=table+8*(body_hash(key)%capacity);node=number(sentinel)
for _ in range(4096):
if not node or node==sentinel:return 0
if number(node+8)==key and (not system or number(node+16,'<B')==0):return number(node+payload)
node=number(node)
return 0
def affine(at):return [struct.unpack('<3d',read(at+i,24)) for i in (0,0x20,0x40,0x60)]
def rotate(v,m):return tuple(sum(v[j]*m[j][i] for j in range(3)) for i in range(3))
def compose(a,b):return [rotate(v,b) for v in a[:3]]+[tuple(x+y for x,y in zip(rotate(a[3],b),b[3]))]
service=int(next(r['address'] for r in sources if r['name']=='spaceBodies'),16);owner=service-0x58
ship_pose=int(pose_lookup['poseAddress'],16);ship_world=number(ship_pose+0xE8)
resource=body_entry(number(owner+0x278),number(owner+0x280),ship_world&0x007fffffffffffff,24,True)
def pose_position(pose):
table=number(pose)
if table==base+0x553DF90 and number(table+0x60)==base+0x3CBC0E0:universal=pose+0xA8
elif number(table+0x60)==base+0x2D9C2C0:universal=pose+0xE8
else:raise ValueError('Unverified original universal pose implementation')
world=number(universal);body=body_entry(number(resource+0x140),number(resource+0x148),world,16)
if not body or number(body+0x378)!=resource:raise ValueError('ATC pose body/resource mismatch')
index=number(body+0x384,'<I');transforms=number(resource+0x388)
if index>=4096:raise ValueError('ATC pose transform index exceeds bounds')
result=affine(universal+0x10)
if number(universal+0x90,'<B'):result=compose(result,affine(transforms+index*0x180+0x80))
return world,compose(result,affine(transforms+index*0x180+0x100))[3]
checks=[]
for item in directory.get('stationMatches',[]):
check=dict(item)
try:
station_root=number(int(item['life'],16)+0xC8)-0x30
station_bound=[]
for label,id_rva in [('maintenanceOperations',0x5F7E800),('servicePermissions',0x5F05CA4),('journalDispatcher',0x5F2DFEC),('marketServicesData',0x5F44D10),('stationData',0x5F05ACC)]:
bound=registered(station_root,number(base+id_rva,'<I'))
if bound:
table=number(bound)
station_bound.append(dict(name=label,address=hex(bound),table=hex(table-base),header=read(bound,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(32)]]))
check['boundServices']=station_bound
station_service=registered(station_root,number(base+0x5F771D0,'<I'))
if station_service:
service_table=number(station_service)
check['stationServices']=dict(address=hex(station_service),table=hex(service_table-base),header=read(station_service,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(service_table+i*8) for i in range(48)]])
station_pose=registered(number(int(item['life'],16)+0xC8)-0x30,pose_lookup['poseTypeId'])
station_table=number(station_pose)
check.update(stationPose=hex(station_pose),table=hex(station_table-base),resource=hex(resource),
stationWorld=str(number(station_pose+0xE8)),shipWorld=str(ship_world),poseHeader=read(station_pose,0x200).hex(),
methods=[hex(number(station_table+i*8)-base) for i in range(14)])
world,position=pose_position(station_pose);_,ship_position=pose_position(ship_pose)
check.update(stationWorld=str(world),shipWorld=str(ship_world),sameSystem=(world&0x007fffffffffffff)==(ship_world&0x007fffffffffffff),
actualDistance=math.dist(position,ship_position),stationPosition=position,shipPosition=ship_position)
except (OSError,ValueError) as error:check['failure']=str(error)
checks.append(check)
report['atcStationGeometry']=checks
service=int(next(r['address'] for r in sources if r['name']=='cruiseTargetService'),16)
targets=[]
for offset in (0x2E0,0x2F0,0x300):
candidate=number(service+offset);item=dict(offset=hex(offset),life=hex(candidate))
try:
if candidate and number(candidate+0xC0,'<I')>4:
item.update(entityId=str(number(candidate+0xB0)),type=number(candidate+0xC0,'<I'))
target_root=number(candidate+0xC8)-0x30;meta=number(target_root+0x10);slots=number(meta+0x30);dense_count=number(meta+0x18)
kind=number(base+0x5F976B0,'<I');slot=number(slots+kind*16)
if 1<=dense_count<=1024 and slot<dense_count:
component=number(number(target_root+0x18)+slot*8);table=number(component)
if in_image(table):item['dropComponent']=dict(address=hex(component),table=hex(table-base),header=read(component,0x600).hex(),methods=[hex(v-base) if in_image(v) else None for v in [number(table+i*8) for i in range(32)]])
except (OSError,ValueError):pass
targets.append(item)
report['cruiseTargetReferences']=dict(mode=number(service+0x2F8,'<I'),targets=targets)
if pose_lookup.get('root'):
root=int(pose_lookup['root'],16);meta=number(root+0x10);dense=number(meta+0x18);slots=number(meta+0x30)
slot=number(slots+number(base+0x5F2F274,'<I')*16)
if 0<dense<=1024 and slot<dense:
loadout=number(number(root+0x18)+slot*8);table=number(loadout)
report['originalAmmoLoadout']={'address':hex(loadout),'table':hex(table-base),'header':read(loadout,0x400).hex(),
'methods':[hex(number(table+i*8)-base) for i in range(40)]}
ammo=[];count=number(loadout+0x1D0);rows=number(loadout+0x1D8)
if count<=128:
for index in range(count):
life=number(rows+index*0x30)
if not life:continue
try:
module_root=number(life+0xC8)-0x30;module_meta=number(module_root+0x10);n=number(module_meta+0x18)
module_slots=number(module_meta+0x30);components=[]
for label,rva in [('clip',0x5F44D80),('hopper',0x5F53A7C)]:
kind=number(base+rva,'<I');dense_index=number(module_slots+kind*16)
if not 0<n<=1024 or dense_index>=n:continue
component=number(number(module_root+0x18)+dense_index*8);vt=number(component)
if in_image(vt):components.append({'kind':label,'address':hex(component),'table':hex(vt-base),'header':read(component,0x100).hex(),
'methods':[hex(number(vt+j*8)-base) for j in range(12)]})
if components:ammo.append({'index':index,'components':components})
except (OSError,ValueError):continue
report['originalAmmoModules']=ammo
args.output.parent.mkdir(parents=True, exist_ok=True)
with args.output.open('x', encoding='utf-8') as output:
json.dump(report, output, indent=2)
print(json.dumps({'player': components[0]['entityId'], 'components': [row['name'] for row in components], 'physicsBindings': len(physics),
'poseLookup': {k: v for k, v in pose_lookup.items() if k not in ('metadataHeader', 'rootHeader')}, 'gameInputsSent': 0, 'screenUse': False}))
finally:
kernel.CloseHandle(process)
if __name__ == '__main__':
main()