XFEEDToolBox

"""Read-only native observation of the receipt-owned offline test process."""
import ctypes as c
from ctypes import wintypes as w
import hashlib
import json
import pathlib
import struct
import sys

PIN = 'e6be8bbe04e6a7ae226d4318945af7f367de13dc5a007a261964d9ba8144e988'
IMAGE = pathlib.Path(r'E:\Elite Dangerous\Products\elite-dangerous-odyssey-64\EliteDangerous64.exe')


class Module(c.Structure):
    _fields_ = [('size', w.DWORD), ('id', w.DWORD), ('pid', w.DWORD),
                ('globalUsage', w.DWORD), ('processUsage', w.DWORD),
                ('base', c.c_void_p), ('bytes', w.DWORD), ('handle', w.HMODULE),
                ('name', w.WCHAR * 256), ('path', w.WCHAR * 260)]


class OwnedNativeProcess:
    def __init__(self, receipt_path):
        self.receipt_path = pathlib.Path(receipt_path).resolve(strict=True)
        self.receipt = json.loads(self.receipt_path.read_text(encoding='utf-8-sig'))
        self.handle = None
        self.lease = None
        if (self.receipt.get('mode') != 'in-process-offline'
                or not self.receipt.get('zeroNetworkCapabilities')
                or self.receipt.get('online')
                or self.receipt.get('imageSha256') != PIN):
            raise ValueError('An offline-only pinned test launch is required')
        script_root = self.receipt_path.parents[2] / 'scripts' / 'capture'
        sys.path.insert(0, str(script_root))
        from owned_diagnostic_lock import acquire_native_lease
        self.lease = acquire_native_lease(self.receipt_path)
        try:
            self.pid = int(self.receipt['ProcessId'])
            self.kernel = k = c.WinDLL('kernel32', use_last_error=True)
            k.OpenProcess.argtypes = [w.DWORD, w.BOOL, w.DWORD]
            k.OpenProcess.restype = w.HANDLE
            k.CloseHandle.argtypes = [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.CreateToolhelp32Snapshot.argtypes = [w.DWORD, w.DWORD]
            k.CreateToolhelp32Snapshot.restype = w.HANDLE
            k.Module32FirstW.argtypes = k.Module32NextW.argtypes = [w.HANDLE, c.POINTER(Module)]
            self.handle = k.OpenProcess(0x410, False, self.pid)
            if not self.handle:
                raise c.WinError(c.get_last_error())
            image = c.create_unicode_buffer(32768)
            length = w.DWORD(len(image))
            if not k.QueryFullProcessImageNameW(self.handle, 0, image, c.byref(length)):
                raise c.WinError(c.get_last_error())
            if pathlib.Path(image.value) != IMAGE or hashlib.sha256(IMAGE.read_bytes()).hexdigest() != PIN:
                raise ValueError('The exact offline test image is not running')
            snapshot = k.CreateToolhelp32Snapshot(0x18, self.pid)
            if snapshot == c.c_void_p(-1).value:
                raise c.WinError(c.get_last_error())
            self.modules = []
            try:
                entry = Module()
                entry.size = c.sizeof(entry)
                found = k.Module32FirstW(snapshot, c.byref(entry))
                while found:
                    self.modules.append((pathlib.Path(entry.path), entry.base, entry.bytes))
                    found = k.Module32NextW(snapshot, c.byref(entry))
            finally:
                k.CloseHandle(snapshot)
            runtime = pathlib.Path(self.receipt['ownedNativeRuntime']).resolve(strict=True)
            if not any(path == runtime for path, _, _ in self.modules):
                raise ValueError('The receipt-owned runtime is absent')
            self.base, self.image_size = next((base, size) for path, base, size in self.modules if path == IMAGE)
        except BaseException:
            self.close()
            raise

    def close(self):
        if self.handle:
            self.kernel.CloseHandle(self.handle)
            self.handle = None
        if self.lease:
            self.lease.close()
            self.lease = None

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

    def read(self, address, size):
        if address < 0x10000 or not 0 < size <= 8192:
            raise ValueError('Invalid bounded memory read')
        buffer = c.create_string_buffer(size)
        done = c.c_size_t()
        if not self.kernel.ReadProcessMemory(self.handle, address, buffer, size, c.byref(done)) or done.value != size:
            raise OSError('The observed object is no longer readable')
        return buffer.raw

    def value(self, address, fmt):
        return struct.unpack(fmt, self.read(address, struct.calcsize(fmt)))[0]

    def ptr(self, address):
        return self.value(address, '<Q')

    def rva(self, address):
        return hex(address - self.base) if self.base <= address < self.base + self.image_size else None

    def ship_seeds(self):
        seeds = {}
        for path in self.receipt_path.parent.glob('entity-*.jsonl'):
            if path.stat().st_size > 32 * 1024 * 1024:
                continue
            for line in path.read_text(encoding='utf-8-sig').splitlines():
                try:
                    row = json.loads(line)
                    if row.get('kind') == 'native-entity-observed' and row.get('type') == 'SideWinder':
                        seeds[row['id']] = int(row['lifecyclePointer'], 16)
                except (ValueError, KeyError):
                    pass
        return seeds