"""Read the pinned world's existing orbital matrices with QUERY_INFORMATION | VM_READ.
No injection, native calls, inputs, or process writes. The initial component
snapshot must refer to the same live player; all body reads are bounded.
"""
import argparse, ctypes as c, ctypes.wintypes as w, hashlib, json, math, struct
from datetime import datetime, timezone
from pathlib import Path
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('snapshot', type=Path)
p.add_argument('output', type=Path)
a = p.parse_args()
d = json.loads(a.snapshot.read_text(encoding='utf-8-sig'))
k = c.WinDLL('kernel32', use_last_error=True)
k.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]; k.OpenProcess.restype = w.HANDLE
k.ReadProcessMemory.argtypes = [w.HANDLE,c.c_void_p,c.c_void_p,c.c_size_t,c.POINTER(c.c_size_t)]
k.QueryFullProcessImageNameW.argtypes = [w.HANDLE,w.DWORD,w.LPWSTR,c.POINTER(w.DWORD)]
k.CloseHandle.argtypes = [w.HANDLE]
h = k.OpenProcess(0x410,False,d['pid'])
if not h: raise c.WinError(c.get_last_error())
try:
name=c.create_unicode_buffer(32768); size=w.DWORD(len(name))
if not k.QueryFullProcessImageNameW(h,0,name,c.byref(size)): raise c.WinError(c.get_last_error())
if hashlib.sha256(Path(name.value).read_bytes()).hexdigest()!=d['gameSha256']: raise ValueError('Game build changed')
base=int(d['base'],16)
def read(at,n):
if at<0x10000 or not 0<n<=8192: raise ValueError('Read outside bound')
buf=c.create_string_buffer(n); done=c.c_size_t()
if not k.ReadProcessMemory(h,at,buf,n,c.byref(done)) or done.value!=n: raise c.WinError(c.get_last_error())
return buf.raw
def u(at,n=8): return int.from_bytes(read(at,n),'little')
def f(at): return struct.unpack('<f',read(at,4))[0]
fsd=int(next(x for x in d['components'] if x['name']=='fsd')['address'],16)
life=int(next(x for x in d['components'] if x['name']=='fsd')['lifecycle'],16)
if u(life+0xc0,4)!=8 or str(u(life+0xb0))!=d['player']['entityId']: raise ValueError('Player identity changed')
pose=int(d['poseLookup']['poseAddress'],16); bodies=u(fsd+0x6a0)
if u(u(bodies)+0x60)!=base+0x3ceb400: raise ValueError('Body service binding changed')
world=u(pose+0xe8); system=world&0x7fffffffffffff
def bucket_key(value):
mask=0xffffffffffffffff
value=(value*0x40000+(~value&mask))&mask
value=((value>>31)^value)*21&mask
value=((value>>11)^value)*65&mask
return ((value>>22)^value)&0xffffffff
def lookup(table_at,capacity_at,key,payload,system_table=False):
table,capacity=u(table_at),u(capacity_at)
if not table or not 0<capacity<=65536: raise ValueError('Body map bounds invalid')
sentinel=table+8*(bucket_key(key)%capacity); node=u(sentinel)
seen=set()
while node!=sentinel:
if node in seen or len(seen)>=4096: raise ValueError('Body map chain invalid')
seen.add(node)
if u(node+8)==key and (not system_table or u(node+16,1)==0): return u(node+payload)
node=u(node)
return 0
# Service is the +0x58 interface of its owner; the original lookup masks
# the system key before borrowing its loaded body-resource map.
owner=bodies-0x58
resource=lookup(owner+0x278,owner+0x280,system,24,True)
if not resource or u(resource+4,4)!=5 or u(resource+0x90,4)==0: raise ValueError('System bodies are not loaded')
def body_for(address): return lookup(resource+0x140,resource+0x148,address,16)
def matrix(at):
values=struct.unpack('<16d',read(at,128))
if not all(math.isfinite(x) for x in values): raise ValueError('Nonfinite matrix')
# Cobra stores three doubles plus an unused alignment lane per row.
# Its affine product treats those lanes as 0/0/0/1, not as live data.
return [list(values[i:i+3])+[1.0 if i==12 else 0.0] for i in range(0,16,4)]
def mul(left,right): return [[sum(left[i][t]*right[t][j] for t in range(4)) for j in range(4)] for i in range(4)]
def global_pose(body,universal=None):
parent=u(body+0x378); index=u(body+0x384,4)
# The loaded transform buffer consists of 0x180-byte slots. Slot
# identity is checked against the body's own immutable address/index.
if parent!=resource or index>=4096: raise ValueError('Body transform owner/index changed')
transforms=u(parent+0x388)
result=matrix(universal+0x10) if universal else [[1.0 if i==j else 0.0 for j in range(4)] for i in range(4)]
if universal and u(universal+0x90,1): result=mul(result,matrix(transforms+index*0x180+0x80))
result=mul(result,matrix(transforms+index*0x180+0x100))
return result
ship_body=body_for(world)
if not ship_body or u(ship_body+0x18)!=world: raise ValueError('Reference body changed')
ship_matrix=global_pose(ship_body,pose+0xe8)
rows=[]
settings=u(bodies-8)
names={int(r['worldAddress']):r['name'] for r in d['navigationRecords']['records'] if r['type']==1}
body_nodes=[];table,capacity=u(resource+0x140),u(resource+0x148)
if not 0<capacity<=65536: raise ValueError('Body map capacity invalid')
seen=set()
for bucket in range(capacity):
sentinel=table+8*bucket;node=u(sentinel)
while node!=sentinel:
if node in seen or len(seen)>=4096:raise ValueError('Body map is cyclic or exceeds bound')
seen.add(node);body_nodes.append((u(node+8),u(node+16)));node=u(node)
for addr,body in body_nodes:
if not body:continue
if u(body+0x18)!=addr or addr&0x7fffffffffffff!=system:raise ValueError('Body identity mismatch')
kind=u(body+0x20,4)
if kind not in (1,2,7): continue
getter=u(u(body)+0x10)
code=read(getter,16)
if code[:6]==bytes.fromhex('488b0148ff60'):
getter=u(u(body)+code[6]); code=read(getter,16)
# Record the pinned leaf getter for static verification. No call.
radius=None
if code[:4]==bytes.fromhex('f30f1081'): radius=f(body+struct.unpack_from('<i',code,4)[0])
elif code[:4]==bytes.fromhex('f30f1041'): radius=f(body+struct.unpack_from('<b',code,4)[0])
elif getter==base+0x3cf2750:
radius=struct.unpack('<f',struct.pack('<f',struct.unpack('<d',read(body+0x710,8))[0]))[0]
entry={'name':names.get(addr,str(addr)),'address':str(addr),'kind':kind,'radiusGetterRva':hex(getter-base),
'radiusGetterBytes':code.hex(),'radius':radius,'body':hex(body),
'matrix':global_pose(body)}
if radius is not None:
star_type=u(body+0x448,4) if kind==1 else None
offset=0x2c if star_type==0x29 else 0x34 if star_type in range(0x19,0x29) else 0x3c if kind==1 else 0x44
entry['starType']=star_type
entry['exclusionRadius']=radius*f(settings+offset)+f(settings+offset+4) if kind in (1,2) else radius
rows.append(entry)
if u(pose+0xe8)!=world or u(life+0xb0)!=int(d['player']['entityId']): raise ValueError('Reference changed during read')
result={'pid':d['pid'],'gameSha256':d['gameSha256'],'entityId':d['player']['entityId'],'observedAt':datetime.now(timezone.utc).isoformat(),
'referenceWorldAddress':str(world),'system':str(system),'referenceBodyKind':u(ship_body+0x20,4),
'rawPosition':list(struct.unpack('<3d',read(pose+0x158,24))),'shipMatrix':ship_matrix,
'bodyCount':len(rows),'bodyMapNodes':len(body_nodes),'bodies':rows,'gameInputsSent':0,'nativeFunctionsInvoked':False,'processAccess':'QUERY_INFORMATION | VM_READ'}
a.output.write_text(json.dumps(result,indent=2))
print(json.dumps({key:value for key,value in result.items() if key!='bodies'},indent=2))
finally: k.CloseHandle(h)
"""Read the pinned world's existing orbital matrices with QUERY_INFORMATION | VM_READ.
No injection, native calls, inputs, or process writes. The initial component
snapshot must refer to the same live player; all body reads are bounded.
"""
import argparse, ctypes as c, ctypes.wintypes as w, hashlib, json, math, struct
from datetime import datetime, timezone
from pathlib import Path
p = argparse.ArgumentParser(description=__doc__)
p.add_argument('snapshot', type=Path)
p.add_argument('output', type=Path)
a = p.parse_args()
d = json.loads(a.snapshot.read_text(encoding='utf-8-sig'))
k = c.WinDLL('kernel32', use_last_error=True)
k.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]; k.OpenProcess.restype = w.HANDLE
k.ReadProcessMemory.argtypes = [w.HANDLE,c.c_void_p,c.c_void_p,c.c_size_t,c.POINTER(c.c_size_t)]
k.QueryFullProcessImageNameW.argtypes = [w.HANDLE,w.DWORD,w.LPWSTR,c.POINTER(w.DWORD)]
k.CloseHandle.argtypes = [w.HANDLE]
h = k.OpenProcess(0x410,False,d['pid'])
if not h: raise c.WinError(c.get_last_error())
try:
name=c.create_unicode_buffer(32768); size=w.DWORD(len(name))
if not k.QueryFullProcessImageNameW(h,0,name,c.byref(size)): raise c.WinError(c.get_last_error())
if hashlib.sha256(Path(name.value).read_bytes()).hexdigest()!=d['gameSha256']: raise ValueError('Game build changed')
base=int(d['base'],16)
def read(at,n):
if at<0x10000 or not 0<n<=8192: raise ValueError('Read outside bound')
buf=c.create_string_buffer(n); done=c.c_size_t()
if not k.ReadProcessMemory(h,at,buf,n,c.byref(done)) or done.value!=n: raise c.WinError(c.get_last_error())
return buf.raw
def u(at,n=8): return int.from_bytes(read(at,n),'little')
def f(at): return struct.unpack('<f',read(at,4))[0]
fsd=int(next(x for x in d['components'] if x['name']=='fsd')['address'],16)
life=int(next(x for x in d['components'] if x['name']=='fsd')['lifecycle'],16)
if u(life+0xc0,4)!=8 or str(u(life+0xb0))!=d['player']['entityId']: raise ValueError('Player identity changed')
pose=int(d['poseLookup']['poseAddress'],16); bodies=u(fsd+0x6a0)
if u(u(bodies)+0x60)!=base+0x3ceb400: raise ValueError('Body service binding changed')
world=u(pose+0xe8); system=world&0x7fffffffffffff
def bucket_key(value):
mask=0xffffffffffffffff
value=(value*0x40000+(~value&mask))&mask
value=((value>>31)^value)*21&mask
value=((value>>11)^value)*65&mask
return ((value>>22)^value)&0xffffffff
def lookup(table_at,capacity_at,key,payload,system_table=False):
table,capacity=u(table_at),u(capacity_at)
if not table or not 0<capacity<=65536: raise ValueError('Body map bounds invalid')
sentinel=table+8*(bucket_key(key)%capacity); node=u(sentinel)
seen=set()
while node!=sentinel:
if node in seen or len(seen)>=4096: raise ValueError('Body map chain invalid')
seen.add(node)
if u(node+8)==key and (not system_table or u(node+16,1)==0): return u(node+payload)
node=u(node)
return 0
# Service is the +0x58 interface of its owner; the original lookup masks
# the system key before borrowing its loaded body-resource map.
owner=bodies-0x58
resource=lookup(owner+0x278,owner+0x280,system,24,True)
if not resource or u(resource+4,4)!=5 or u(resource+0x90,4)==0: raise ValueError('System bodies are not loaded')
def body_for(address): return lookup(resource+0x140,resource+0x148,address,16)
def matrix(at):
values=struct.unpack('<16d',read(at,128))
if not all(math.isfinite(x) for x in values): raise ValueError('Nonfinite matrix')
# Cobra stores three doubles plus an unused alignment lane per row.
# Its affine product treats those lanes as 0/0/0/1, not as live data.
return [list(values[i:i+3])+[1.0 if i==12 else 0.0] for i in range(0,16,4)]
def mul(left,right): return [[sum(left[i][t]*right[t][j] for t in range(4)) for j in range(4)] for i in range(4)]
def global_pose(body,universal=None):
parent=u(body+0x378); index=u(body+0x384,4)
# The loaded transform buffer consists of 0x180-byte slots. Slot
# identity is checked against the body's own immutable address/index.
if parent!=resource or index>=4096: raise ValueError('Body transform owner/index changed')
transforms=u(parent+0x388)
result=matrix(universal+0x10) if universal else [[1.0 if i==j else 0.0 for j in range(4)] for i in range(4)]
if universal and u(universal+0x90,1): result=mul(result,matrix(transforms+index*0x180+0x80))
result=mul(result,matrix(transforms+index*0x180+0x100))
return result
ship_body=body_for(world)
if not ship_body or u(ship_body+0x18)!=world: raise ValueError('Reference body changed')
ship_matrix=global_pose(ship_body,pose+0xe8)
rows=[]
settings=u(bodies-8)
names={int(r['worldAddress']):r['name'] for r in d['navigationRecords']['records'] if r['type']==1}
body_nodes=[];table,capacity=u(resource+0x140),u(resource+0x148)
if not 0<capacity<=65536: raise ValueError('Body map capacity invalid')
seen=set()
for bucket in range(capacity):
sentinel=table+8*bucket;node=u(sentinel)
while node!=sentinel:
if node in seen or len(seen)>=4096:raise ValueError('Body map is cyclic or exceeds bound')
seen.add(node);body_nodes.append((u(node+8),u(node+16)));node=u(node)
for addr,body in body_nodes:
if not body:continue
if u(body+0x18)!=addr or addr&0x7fffffffffffff!=system:raise ValueError('Body identity mismatch')
kind=u(body+0x20,4)
if kind not in (1,2,7): continue
getter=u(u(body)+0x10)
code=read(getter,16)
if code[:6]==bytes.fromhex('488b0148ff60'):
getter=u(u(body)+code[6]); code=read(getter,16)
# Record the pinned leaf getter for static verification. No call.
radius=None
if code[:4]==bytes.fromhex('f30f1081'): radius=f(body+struct.unpack_from('<i',code,4)[0])
elif code[:4]==bytes.fromhex('f30f1041'): radius=f(body+struct.unpack_from('<b',code,4)[0])
elif getter==base+0x3cf2750:
radius=struct.unpack('<f',struct.pack('<f',struct.unpack('<d',read(body+0x710,8))[0]))[0]
entry={'name':names.get(addr,str(addr)),'address':str(addr),'kind':kind,'radiusGetterRva':hex(getter-base),
'radiusGetterBytes':code.hex(),'radius':radius,'body':hex(body),
'matrix':global_pose(body)}
if radius is not None:
star_type=u(body+0x448,4) if kind==1 else None
offset=0x2c if star_type==0x29 else 0x34 if star_type in range(0x19,0x29) else 0x3c if kind==1 else 0x44
entry['starType']=star_type
entry['exclusionRadius']=radius*f(settings+offset)+f(settings+offset+4) if kind in (1,2) else radius
rows.append(entry)
if u(pose+0xe8)!=world or u(life+0xb0)!=int(d['player']['entityId']): raise ValueError('Reference changed during read')
result={'pid':d['pid'],'gameSha256':d['gameSha256'],'entityId':d['player']['entityId'],'observedAt':datetime.now(timezone.utc).isoformat(),
'referenceWorldAddress':str(world),'system':str(system),'referenceBodyKind':u(ship_body+0x20,4),
'rawPosition':list(struct.unpack('<3d',read(pose+0x158,24))),'shipMatrix':ship_matrix,
'bodyCount':len(rows),'bodyMapNodes':len(body_nodes),'bodies':rows,'gameInputsSent':0,'nativeFunctionsInvoked':False,'processAccess':'QUERY_INFORMATION | VM_READ'}
a.output.write_text(json.dumps(result,indent=2))
print(json.dumps({key:value for key,value in result.items() if key!='bodies'},indent=2))
finally: k.CloseHandle(h)
