OpenFUT/fifa17-recon/tools/offline_match_locator.py

#!/usr/bin/env python3
"""Read-only dynamic locator for FIFA17 Offline Seasons match state.

Never relies on heap addresses or allocator handles. It identifies:

* the 10 x 16-byte parsed fixture array from its complete wire-derived record
  sequence (teamId/difficulty/roundId/rewardMult/coins),
* match-team records from the corrected invariant prefix (11,7,0,0,76), never
  from the transient +0x18 handle,
* the match-config team pair from structural fields around it, not its team ids.

    offline_match_locator.py [pid] [--fixture-index 0] [--json]
    offline_match_locator.py --selftest

READ-ONLY: /proc/<pid>/mem is opened 'rb'. No debugger and no game input.
"""
from __future__ import annotations

import argparse
import glob
import json
import os
import re
import struct
import sys
from dataclasses import asdict, dataclass

DEFAULT_TEAMS = (73, 240, 241, 243, 73, 240, 241, 243, 73, 240)
MATCH_HEADER = struct.pack("<5i", 11, 7, 0, 0, 76)
PARTICIPANT_PREFIX = struct.pack("<8i", -1, -2, -1, -2, -1, -2, -1, -2)
F01 = 0x3DCCCCCD


@dataclass
class Fixture:
    address: int
    selected_address: int
    selected_index: int
    selected_team_id: int
    records: list[dict[str, int]]


@dataclass
class MatchTeam:
    address: int
    team_id: int
    marker_18: int
    marker_1c: int
    xi: list[int]
    substitutes: list[int]


@dataclass
class MatchConfig:
    pair_address: int
    team_id_0: int
    team_id_1: int
    player_count_0: int
    player_count_1: int


def find_pids() -> list[int]:
    """All live FIFA17.exe processes, largest resident set first.

    The UMU/Proton launch chain briefly creates a small process with the same
    comm before the real game. Returning the first /proc glob match attached
    the trace supervisor to that short-lived process and missed the match.
    """
    found = []
    for directory in glob.glob("/proc/[0-9]*"):
        try:
            with open(os.path.join(directory, "comm")) as handle:
                if handle.read().strip() != "FIFA17.exe":
                    continue
            pid = int(os.path.basename(directory))
            with open(os.path.join(directory, "statm")) as handle:
                resident_pages = int(handle.read().split()[1])
            found.append((resident_pages, pid))
        except (OSError, ValueError, IndexError):
            continue
    return [pid for _resident, pid in sorted(found, reverse=True)]


def find_pid() -> int | None:
    pids = find_pids()
    return pids[0] if pids else None


def fixture_bytes(teams: tuple[int, ...] = DEFAULT_TEAMS) -> bytes:
    return b"".join(
        struct.pack("<iBBHii", team_id, 1, round_id, 0, 1, 400)
        for round_id, team_id in enumerate(teams)
    )


def readable_regions(pid: int, *, writable_anon_only: bool = False):
    with open(f"/proc/{pid}/maps") as maps:
        for line in maps:
            match = re.match(
                r"([0-9a-f]+)-([0-9a-f]+)\s+(\S{4})\s+\S+\s+\S+\s+\S+\s*(.*)",
                line,
            )
            if not match:
                continue
            lo, hi = int(match.group(1), 16), int(match.group(2), 16)
            perms, path = match.group(3), match.group(4).strip()
            if perms[0] != "r" or path.startswith(("/dev", "/memfd")):
                continue
            if hi - lo > 512 * 1024 * 1024:
                continue
            if writable_anon_only and (perms[1] != "w" or path):
                continue
            yield lo, hi, perms, path


def _i32(buf: bytes, offset: int) -> int:
    return struct.unpack_from("<i", buf, offset)[0]


def scan_fixture_buffer(buf: bytes, base: int, selected_index: int) -> list[Fixture]:
    pattern = fixture_bytes()
    found = []
    offset = buf.find(pattern)
    while offset >= 0:
        records = []
        for round_id in range(len(DEFAULT_TEAMS)):
            at = offset + round_id * 16
            team_id, difficulty, parsed_round, _pad, reward_mult, coins = struct.unpack_from(
                "<iBBHii", buf, at
            )
            records.append(
                {
                    "team_id": team_id,
                    "difficulty": difficulty,
                    "round_id": parsed_round,
                    "reward_mult": reward_mult,
                    "coins": coins,
                }
            )
        found.append(
            Fixture(
                address=base + offset,
                selected_address=base + offset + selected_index * 16,
                selected_index=selected_index,
                selected_team_id=records[selected_index]["team_id"],
                records=records,
            )
        )
        offset = buf.find(pattern, offset + 4)
    return found


def scan_match_team_buffer(buf: bytes, base: int) -> list[MatchTeam]:
    found = []
    offset = buf.find(MATCH_HEADER)
    while offset >= 0:
        if offset + 0x7C <= len(buf):
            found.append(
                MatchTeam(
                    address=base + offset,
                    team_id=_i32(buf, offset + 0x14),
                    marker_18=_i32(buf, offset + 0x18),
                    marker_1c=_i32(buf, offset + 0x1C),
                    xi=list(struct.unpack_from("<11i", buf, offset + 0x20)),
                    substitutes=list(struct.unpack_from("<12i", buf, offset + 0x4C)),
                )
            )
        offset = buf.find(MATCH_HEADER, offset + 4)
    return found


def _valid_config(buf: bytes, pair: int) -> bool:
    required = pair + 0x50
    if pair < 0 or required > len(buf):
        return False
    return (
        tuple(struct.unpack_from("<4I", buf, pair + 0x1C)) == (F01, F01, F01, F01)
        and _i32(buf, pair + 0x38) == 11
        and _i32(buf, pair + 0x3C) == 11
        and _i32(buf, pair + 0x40) == 0
        and _i32(buf, pair + 0x44) == 5
    )


def scan_match_config_buffer(buf: bytes, base: int) -> list[MatchConfig]:
    found = []
    offset = buf.find(PARTICIPANT_PREFIX)
    while offset >= 0:
        pair = offset + len(PARTICIPANT_PREFIX)
        if _valid_config(buf, pair):
            found.append(
                MatchConfig(
                    pair_address=base + pair,
                    team_id_0=_i32(buf, pair),
                    team_id_1=_i32(buf, pair + 4),
                    player_count_0=_i32(buf, pair + 0x38),
                    player_count_1=_i32(buf, pair + 0x3C),
                )
            )
        offset = buf.find(PARTICIPANT_PREFIX, offset + 4)
    return found


def scan_process(
    pid: int,
    selected_index: int,
    *,
    include_fixture: bool = True,
    writable_anon_only: bool = False,
) -> dict[str, list]:
    result: dict[str, list] = {"fixtures": [], "match_teams": [], "match_configs": []}
    with open(f"/proc/{pid}/mem", "rb", 0) as memory:
        for lo, hi, _perms, _path in readable_regions(
            pid, writable_anon_only=writable_anon_only
        ):
            try:
                memory.seek(lo)
                buf = memory.read(hi - lo)
            except (OSError, ValueError, OverflowError):
                continue
            if include_fixture:
                result["fixtures"].extend(scan_fixture_buffer(buf, lo, selected_index))
            result["match_teams"].extend(scan_match_team_buffer(buf, lo))
            result["match_configs"].extend(scan_match_config_buffer(buf, lo))
    return result


def selftest() -> None:
    fixture = fixture_bytes()
    team = bytearray(0x7C)
    team[:20] = MATCH_HEADER
    struct.pack_into("<iii", team, 0x14, 130000, 0x54001, 0x54002)
    struct.pack_into("<11i", team, 0x20, *range(11))
    struct.pack_into("<12i", team, 0x4C, *range(20, 32))
    config = bytearray(0x20 + 0x50)
    config[:0x20] = PARTICIPANT_PREFIX
    pair = 0x20
    struct.pack_into("<ii", config, pair, 130000, 130000)
    struct.pack_into("<4I", config, pair + 0x1C, F01, F01, F01, F01)
    struct.pack_into("<iiii", config, pair + 0x38, 11, 11, 0, 5)
    buf = b"X" * 32 + fixture + b"Y" * 32 + team + b"Z" * 32 + config
    fixtures = scan_fixture_buffer(buf, 0x1000, 0)
    teams = scan_match_team_buffer(buf, 0x1000)
    configs = scan_match_config_buffer(buf, 0x1000)
    assert len(fixtures) == 1 and fixtures[0].selected_team_id == 73
    assert len(teams) == 1 and teams[0].team_id == 130000
    assert len(configs) == 1 and configs[0].team_id_1 == 130000
    # The transient handle is never part of the anchor.
    struct.pack_into("<i", team, 0x18, -1)
    assert len(scan_match_team_buffer(bytes(team), 0)) == 1
    print("offline_match_locator selftest: PASS")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("pid", nargs="?", type=int)
    parser.add_argument("--fixture-index", type=int, default=0)
    parser.add_argument("--json", action="store_true")
    parser.add_argument("--selftest", action="store_true")
    parser.add_argument("--writable-anon-only", action="store_true")
    args = parser.parse_args()
    if args.selftest:
        selftest()
        return 0
    pid = args.pid or find_pid()
    if not pid:
        print("FIFA17.exe not found", file=sys.stderr)
        return 2
    if not 0 <= args.fixture_index < len(DEFAULT_TEAMS):
        print("--fixture-index must be 0..9", file=sys.stderr)
        return 2
    result = scan_process(
        pid,
        args.fixture_index,
        writable_anon_only=args.writable_anon_only,
    )
    serial = {key: [asdict(value) for value in values] for key, values in result.items()}
    serial["pid"] = pid
    if args.json:
        print(json.dumps(serial, sort_keys=True))
        return 0
    print(f"pid={pid}")
    for fixture in result["fixtures"]:
        print(
            f"fixture @0x{fixture.address:x}; selected index {fixture.selected_index} "
            f"@0x{fixture.selected_address:x} teamId={fixture.selected_team_id}"
        )
    for config in result["match_configs"]:
        print(
            f"match config pair @0x{config.pair_address:x}: "
            f"[{config.team_id_0}, {config.team_id_1}]"
        )
    for team in result["match_teams"]:
        print(
            f"match team @0x{team.address:x}: teamId={team.team_id} "
            f"handles=[{team.marker_18}, {team.marker_1c}]"
        )
    print(
        f"counts: fixtures={len(result['fixtures'])} "
        f"configs={len(result['match_configs'])} teams={len(result['match_teams'])}"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())