OpenFUT/fifa17-recon/tools/live/watch_residency.py

#!/usr/bin/env python3
"""Watch FIFA 17's resident club-item store and log every change, with timestamps.

Read-only. Waits for FIFA17.exe to appear, re-resolves the store each tick (the
manager is reallocated across logins), and appends one line per CHANGE so the
output can be aligned against the staging host's route log by wall clock.

Purpose: answer "after which response does a resident club item first appear?"
without reversing the constructor first. Pair with

    journalctl -u openfut-staging-host --since <start> -o short-iso

and compare timestamps.

Usage:  watch_residency.py [--interval 1.0] [--out /path/log] [--once]
"""
from __future__ import annotations

import argparse
import collections
import os
import re
import struct
import sys
import time

CARDS_DLL = "CardsDLL_Win64_retail.dll"
OWNER_GLOBAL = 0x1802E6398        # FUN_18011a830: mov rax,[this]; ret
SANITY_VA = 0x180026FEA           # mov edx,0x7575
SANITY_BYTES = bytes.fromhex("ba75750000")
IMAGE_BASE = 0x180000000

# item-record offsets, all previously proven (see Vault: Kit Selector APT Decode)
OFF = {"cardtype": 0x4C, "cardsubtypeid": 0x50, "itemState": 0x5C,
       "category": 0x60, "teamid": 0x94}
OFF_KITTYPE_U16 = 0xBA


class Target:
    """One live FIFA17.exe, with the store chain resolved."""

    def __init__(self, pid: int):
        self.pid = pid
        self.mem = open(f"/proc/{pid}/mem", "rb", buffering=0)
        self.base = self._cards_base()
        if self.base is None:
            raise RuntimeError("CardsDLL mapping not found")
        probe = self.rd(self.live(SANITY_VA), 5)
        if probe != SANITY_BYTES:
            raise RuntimeError(f"base sanity failed: {probe.hex(' ')}")
        owner = self.q(self.live(OWNER_GLOBAL))
        if not owner:
            raise RuntimeError("owner object is null (not logged in yet)")
        vt = self.q(owner)
        getter = self.q(vt + 0x4E8)
        b = self.rd(getter, 8)
        # lea rax,[rcx+imm32] ; ret   /   lea rax,[rcx+imm8] ; ret
        if b[0:3] == bytes.fromhex("488d81"):
            self.mgr = owner + struct.unpack_from("<I", b, 3)[0]
        elif b[0:3] == bytes.fromhex("488d41"):
            self.mgr = owner + b[3]
        elif b[0:3] == bytes.fromhex("488b81"):
            self.mgr = self.q(owner + struct.unpack_from("<I", b, 3)[0])
        else:
            raise RuntimeError(f"unrecognised getter: {b.hex(' ')}")

    # -- raw access ------------------------------------------------------
    def rd(self, a: int, n: int) -> bytes:
        self.mem.seek(a)
        return self.mem.read(n)

    def q(self, a: int) -> int:
        return struct.unpack("<Q", self.rd(a, 8))[0]

    def live(self, static: int) -> int:
        return self.base + (static - IMAGE_BASE)

    def _cards_base(self):
        named = []
        for ln in open(f"/proc/{self.pid}/maps"):
            m = re.match(r"([0-9a-f]+)-([0-9a-f]+) \S{4} \S+ \S+ \S+\s+(.+)", ln)
            if m:
                named.append((int(m.group(1), 16), m.group(3).strip()))
        # NEAREST PRECEDING NAMED mapping: Wine maps PE sections anonymously and
        # the Wine heap is also rwx, so permissions cannot identify a module.
        for start, path in sorted(named):
            if path.endswith(CARDS_DLL):
                return start
        return None

    # -- the store -------------------------------------------------------
    def vector(self, off_begin: int):
        beg, end = self.q(self.mgr + off_begin), self.q(self.mgr + off_begin + 8)
        if not (0 < beg <= end) or (end - beg) % 24 or (end - beg) > 24 * 200000:
            return None, 0
        return beg, (end - beg) // 24

    def records(self, off_begin: int):
        beg, n = self.vector(off_begin)
        out = []
        if beg is None:
            return out
        for k in range(n):
            try:
                rec = self.q(beg + k * 24 + 0x10)
            except OSError:
                continue
            if not rec:
                out.append(None)
                continue
            try:
                r = self.rd(rec, 0xC0)
            except OSError:
                out.append(None)
                continue
            if len(r) < 0xC0:
                out.append(None)
                continue
            f = {k2: struct.unpack_from("<i", r, v)[0] for k2, v in OFF.items()}
            f["teamkittypetechid"] = struct.unpack_from("<H", r, OFF_KITTYPE_U16)[0]
            f["ptr"] = rec
            out.append(f)
        return out

    def snapshot(self) -> dict:
        club = self.records(0x108)
        players = self.records(0xD8)
        hist = collections.Counter(
            (r["cardtype"], r["cardsubtypeid"]) for r in club if r
        )
        return {
            "club_slots": len(club),
            "club_filled": sum(1 for r in club if r),
            "club_hist": dict(hist),
            "club_records": [r for r in club if r],
            "player_slots": len(players),
            "player_filled": sum(1 for r in players if r),
        }


def find_pid() -> int | None:
    for d in os.listdir("/proc"):
        if not d.isdigit():
            continue
        try:
            with open(f"/proc/{d}/comm") as f:
                if f.read().strip() == "FIFA17.exe":
                    return int(d)
        except OSError:
            continue
    return None


def fmt(snap: dict) -> str:
    parts = [
        f"club={snap['club_filled']}/{snap['club_slots']}",
        f"players={snap['player_filled']}/{snap['player_slots']}",
    ]
    if snap["club_hist"]:
        parts.append("hist=" + ",".join(
            f"(ct{a},st{b})x{c}" for (a, b), c in sorted(snap["club_hist"].items())))
    for r in snap["club_records"]:
        parts.append(
            "KIT[" if (r["cardtype"], r["cardsubtypeid"]) == (7, 9) else "rec[")
        parts[-1] += (f"ptr={r['ptr']:#x} ct={r['cardtype']} st={r['cardsubtypeid']} "
                      f"state={r['itemState']} cat={r['category']} "
                      f"team={r['teamid']} kittype={r['teamkittypetechid']}]")
    return " ".join(parts)


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--interval", type=float, default=1.0)
    ap.add_argument("--out", default="/home/alex/openfut-live/residency.log")
    ap.add_argument("--once", action="store_true")
    a = ap.parse_args()

    sink = sys.stdout if a.out == "-" else open(a.out, "a", buffering=1)

    def emit(msg: str) -> None:
        line = f"{time.strftime('%Y-%m-%dT%H:%M:%S%z')} {msg}"
        print(line, file=sink)
        if sink is not sys.stdout:
            print(line, flush=True)

    emit("watch: start")
    target = None
    last = None
    while True:
        if target is None:
            pid = find_pid()
            if pid is None:
                if a.once:
                    emit("watch: no FIFA17.exe"); return 1
                time.sleep(a.interval); continue
            try:
                target = Target(pid)
                emit(f"watch: attached pid={pid} cardsdll={target.base:#x} "
                     f"mgr={target.mgr:#x}")
                last = None
            except (OSError, RuntimeError) as e:
                # not logged in yet, or the process died mid-resolve
                if a.once:
                    emit(f"watch: not ready: {e}"); return 1
                target = None
                time.sleep(a.interval); continue
        try:
            snap = target.snapshot()
        except (OSError, struct.error) as e:
            emit(f"watch: detached ({e})")
            target = None
            if a.once:
                return 1
            continue
        key = fmt(snap)
        if key != last:
            emit(key)
            last = key
        if a.once:
            return 0
        time.sleep(a.interval)


if __name__ == "__main__":
    sys.exit(main())