XFEEDToolBox

"""Request a game-calculated route through the original player interface; no flight input."""
import argparse
import hashlib
import json
import time
from pathlib import Path

import frida


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('--snapshot', type=Path, required=True)
    parser.add_argument('--target', type=int, required=True)
    parser.add_argument('--journal', type=Path, required=True)
    parser.add_argument('--output', type=Path, required=True)
    args = parser.parse_args()
    profile = json.loads(args.profile.read_text(encoding='utf-8-sig'))
    snapshot = json.loads(args.snapshot.read_text(encoding='utf-8-sig'))
    if (snapshot['pid'] != args.pid or snapshot['gameSha256'] != profile['sha256']
            or hashlib.sha256(Path(profile['image']).read_bytes()).hexdigest() != profile['sha256']
            or profile['sha256'] != '6891f70a021fa9f3a0e00458c284f48d1226e68ae7fc02ad48081aed47857e93'):
        raise ValueError('Selected process/build does not match the route observation')
    player = snapshot['player']
    if not player['inGame'] or player['controlApplied'] or player['fsd']['state'] != 2 or player['fsd']['ownedCharging']:
        raise ValueError('Route-only verification requires an idle player ship without automatic control')
    routes = [row for row in snapshot['routeBindings'] if row['owner'] == 'pilot' and row['name'] == 'route']
    if len(routes) != 1 or routes[0]['typeId'] != 199 or args.target <= 0 or args.target >= 1 << 55:
        raise ValueError('A unique original player route interface and valid system address are required')
    route = routes[0]
    fsd = next(row for row in snapshot['components'] if row['name'] == 'fsd')
    bindings = dict(pid=args.pid, entity=player['entityId'], fsd=fsd['address'], life=fsd['lifecycle'],
                    root=route['root'], route=route['address'], slot=route['denseSlot'],
                    thread=player['fsd']['lastHookThread'], mode=player['rawFsdMode'], target=str(args.target))
    source = 'const PROFILE=' + json.dumps(profile) + ';const BINDINGS=' + json.dumps(bindings) + ';\n'
    source += Path(__file__).with_name('navigation-route-probe.js').read_text(encoding='utf-8')
    route_file = args.journal / 'NavRoute.json'
    before = json.loads(route_file.read_text(encoding='utf-8-sig')) if route_file.exists() else None
    args.output.parent.mkdir(parents=True, exist_ok=True)
    failures, requests = [], []
    with args.output.open('x', encoding='utf-8') as output:
        output.write(json.dumps(dict(kind='route-before', data=before)) + '\n')
        output.flush()
        session = frida.attach(args.pid)
        try:
            script = session.create_script(source)

            def message(msg, data):
                row = msg.get('payload') if msg['type'] == 'send' else dict(kind='error', detail=msg)
                output.write(json.dumps(row) + '\n')
                output.flush()
                if row.get('kind') == 'error':
                    failures.append(row)
                if row.get('kind') == 'original-route-request':
                    requests.append(row)
                print(json.dumps(row), flush=True)

            script.on('message', message)
            script.load()
            deadline = time.monotonic() + 30
            observed = None
            while time.monotonic() < deadline and not failures:
                if route_file.exists():
                    try:
                        observed = json.loads(route_file.read_text(encoding='utf-8-sig'))
                        waypoints = observed.get('Route', [])
                        if requests and waypoints and waypoints[-1].get('SystemAddress') == args.target and observed != before:
                            break
                    except (OSError, ValueError):
                        pass
                time.sleep(.1)
            script.unload()
        finally:
            session.detach()
        waypoints = (observed or {}).get('Route', [])
        passed = len(requests) == 1 and not failures and bool(waypoints) and waypoints[-1].get('SystemAddress') == args.target and observed != before
        result = dict(kind='route-result', passed=passed, requests=len(requests), failures=failures,
                      actualGameRoute=observed, flightInputsSent=0, journalWrites=0, screenUse=False)
        output.write(json.dumps(result) + '\n')
        print(json.dumps(result), flush=True)
    if not passed:
        raise SystemExit(1)


if __name__ == '__main__':
    main()