XFEEDToolBox

"""Bounded native DLL verification. No screen or OS-input automation."""
import argparse
import ctypes as c
from ctypes import wintypes as w
import hashlib
import json
from pathlib import Path
import time
import uuid

p=argparse.ArgumentParser(description=__doc__)
p.add_argument('--pid',type=int,required=True)
p.add_argument('--profile',type=Path,required=True)
p.add_argument('--dll',type=Path)
p.add_argument('--output',type=Path,required=True)
p.add_argument('--seconds',type=int,default=8,choices=range(2,31))
p.add_argument('--axis',choices=['observe','yaw','pitch','forward'],default='observe')
args=p.parse_args()
profile=json.loads(args.profile.read_text(encoding='utf-8-sig'))
if hashlib.sha256(Path(profile['image']).read_bytes()).hexdigest()!=profile['sha256']:
    raise SystemExit('Game image does not match the native build profile')
if args.dll:
    import frida
    dll=args.dll.resolve(strict=True)
    if not dll.is_relative_to(Path('artifacts/navigation-native-build').resolve()):
        raise SystemExit('Only this task\'s own native build may be loaded')
    session=frida.attach(args.pid)
    try:
        script=session.create_script('''
const game=Process.getModuleByName('EliteDangerous64.exe');
if(game.path.toLowerCase()!==IMAGE.toLowerCase())throw new Error('Unexpected game image');
rpc.exports={load(path){const load=new NativeFunction(Module.getGlobalExportByName('LoadLibraryW'),'pointer',['pointer']);const module=load(Memory.allocUtf16String(path));if(module.isNull())throw new Error('Native module load failed');const own=Process.findModuleByAddress(module);const start=new NativeFunction(own.getExportByName('XFENavigationStart'),'uint',[]);if(start()!==1)throw new Error('Explicit native observation initialization failed');return module.toString();}};
'''.replace('IMAGE',json.dumps(profile['image'])))
        script.load();print(json.dumps({'loadedModule':script.exports_sync.load(str(dll)),'sha256':hashlib.sha256(dll.read_bytes()).hexdigest()}),flush=True);script.unload()
    finally:session.detach()

k=c.WinDLL('kernel32',use_last_error=True)
k.CreateFileW.argtypes=[w.LPCWSTR,w.DWORD,w.DWORD,c.c_void_p,w.DWORD,w.DWORD,w.HANDLE];k.CreateFileW.restype=w.HANDLE
k.WaitNamedPipeW.argtypes=[w.LPCWSTR,w.DWORD];k.WaitNamedPipeW.restype=w.BOOL
k.WriteFile.argtypes=[w.HANDLE,c.c_void_p,w.DWORD,c.POINTER(w.DWORD),c.c_void_p];k.WriteFile.restype=w.BOOL
k.ReadFile.argtypes=[w.HANDLE,c.c_void_p,w.DWORD,c.POINTER(w.DWORD),c.c_void_p];k.ReadFile.restype=w.BOOL
k.CloseHandle.argtypes=[w.HANDLE]
pipe_name=rf'\\.\pipe\XFEEDToolBox.Navigation.{args.pid}'
def call(request):
    # The single-instance server closes and recreates its endpoint between messages.
    # Retry only connection establishment; never replay a command after a write.
    deadline=time.monotonic()+.25
    while True:
        if k.WaitNamedPipeW(pipe_name,40):
            pipe=k.CreateFileW(pipe_name,0xC0000000,0,None,3,0,None)
            if pipe!=c.c_void_p(-1).value:break
        error=c.get_last_error()
        if error not in (2,121,231) or time.monotonic()>=deadline:raise c.WinError(error)
        time.sleep(.001)
    try:
        body=(json.dumps(request)+'\n').encode();count=w.DWORD()
        if not k.WriteFile(pipe,body,len(body),c.byref(count),None):raise c.WinError(c.get_last_error())
        buffer=c.create_string_buffer(65536)
        if not k.ReadFile(pipe,buffer,len(buffer),c.byref(count),None):raise c.WinError(c.get_last_error())
        result=json.loads(buffer.raw[:count.value])
        if not result.get('ok'):raise RuntimeError(result.get('error','Native operation failed'))
        return result['result']
    finally:k.CloseHandle(pipe)

args.output.parent.mkdir(parents=True,exist_ok=True)
owner=uuid.uuid4().hex
with args.output.open('x',encoding='utf-8') as output:
    deadline=time.monotonic()+5
    while True:
        try:description=call({'op':'describe'});break
        except OSError:
            if time.monotonic()>deadline:raise
            time.sleep(.15)
    output.write(json.dumps({'kind':'native-description',**description})+'\n');output.flush();print(json.dumps(description),flush=True)
    if description.get('gameSha256')!=profile['sha256'] or description.get('processId')!=args.pid:raise RuntimeError('Native module identity mismatch')
    if not description.get('continuousPose'):
        raise RuntimeError('Native telemetry is not available; no input has been sent')
    begin=time.monotonic();samples=0
    try:
        while time.monotonic()-begin<args.seconds:
            frame=call({'op':'read'});elapsed=time.monotonic()-begin
            row={'kind':'native-pipe-sample','elapsed':elapsed,**frame}
            if args.axis!='observe':
                # A brief low-authority pulse, followed by zero axes; no FSD commands are supported here.
                axes=[0.0]*6
                if 2<=elapsed<4:axes[{'forward':0,'yaw':3,'pitch':4}[args.axis]]=.08
                row['command']=call({'op':'calibrate.axes','owner':owner,'sequence':frame['sequence'],'ttlMilliseconds':180,'axes':axes})
                row['requestedAxes']=axes
            output.write(json.dumps(row)+'\n');output.flush();samples+=1;time.sleep(.045)
    finally:
        if args.axis!='observe':
            release=call({'op':'release','owner':owner});output.write(json.dumps({'kind':'released',**release})+'\n');output.flush()
            for _ in range(8):
                time.sleep(.06)
                output.write(json.dumps({'kind':'native-release-observation',**call({'op':'read'})})+'\n');output.flush()
    print(json.dumps({'samples':samples,'mode':args.axis,'screenUse':False,'fsdCommandsSent':0}),flush=True)