游戏云存档被写坏：反编译游戏代码重建存档并回写云端

#!/usr/bin/env python
"""Read the LOST TRIGGER (PlayFab title 142A3) cloud save for a given in-game ID.

Usage: python playfab_fetch.py <16-char in-game ID>
Writes raw API responses to ./cloud_raw/*.json and prints only metadata
(key names, value lengths, missing-field audit). The PlayFab SessionTicket is
never printed.
"""
import json, os, sys, urllib.request, urllib.error, re

TITLE = "142A3"
OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cloud_raw")
os.makedirs(OUT, exist_ok=True)


def post(path, body, ticket=None):
    url = f"https://{TITLE}.playfabapi.com/{path}"
    data = json.dumps(body).encode()
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Content-Type", "application/json")
    if ticket:
        req.add_header("X-Authorization", ticket)
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"raw": raw}


def main():
    pid = sys.argv[1].strip() if len(sys.argv) > 1 else open("player_id.txt").read().strip()
    print(f"[i] CustomId length = {len(pid)}")

    st, resp = post("Client/LoginWithCustomID",
                    {"TitleId": TITLE, "CustomId": pid, "CreateAccount": False, "InfoRequestParameters": {"GetUserData": False}})
    json.dump(resp, open(os.path.join(OUT, "login.json"), "w"), indent=1)
    if resp.get("code") != 200:
        print("[x] login failed:", resp.get("errorCode"), "-", resp.get("errorMessage"))
        return
    d = resp["data"]
    ticket = d["SessionTicket"]
    print("[+] login ok; PlayFabId:", d.get("PlayFabId"), "| NewlyCreated:", d.get("NewlyCreated"))
    print("[i] SessionTicket length:", len(ticket), "prefix:", ticket[:6] + "…")

    st, resp = post("Client/GetUserData", {}, ticket)
    json.dump(resp, open(os.path.join(OUT, "userdata.json"), "w"), indent=1)
    if resp.get("code") != 200:
        print("[x] GetUserData failed:", resp.get("errorCode"), "-", resp.get("errorMessage"))
        return
    data = resp["data"].get("Data", {})
    print(f"[+] cloud keys: {sorted(data.keys())}")
    outdir = os.path.join(OUT, "values")
    os.makedirs(outdir, exist_ok=True)
    for k, v in sorted(data.items()):
        val = v.get("Value")
        print(f"    {k}: len={len(val) if val is not None else 0} LastUpdated={v.get('LastUpdated')}")
        if val is not None:
            open(os.path.join(outdir, k + ".json"), "w", encoding="utf-8").write(val)

    # ---- audit against the fields the installed version expects ----
    expected = {}
    src = os.path.join(os.path.dirname(os.path.abspath(__file__)),
                       "dump_all/CodeEntries/gml_Object_o_popup_option_Step_0.gml")
    if os.path.exists(src):
        lines = open(src, encoding="utf-8", errors="replace").read().split("\n")
        bounds = {"userdata": (23, 1736), "unitdata": (1749, 2105), "etcdata": (2118, 2895)}
        for name, (a, b) in bounds.items():
            keys = []
            for ln in lines[a - 1:b]:
                m = re.match(r"\s*([A-Za-z_][A-Za-z0-9_]*):\s", ln)
                if m:
                    keys.append(m.group(1))
            expected[name] = keys
    for name, keys in expected.items():
        if name not in data:
            print(f"[!] cloud has NO '{name}' key (expected {len(keys)} fields)")
            continue
        try:
            got = json.loads(data[name]["Value"])
        except Exception as e:
            print(f"[!] '{name}' is not valid JSON: {e}")
            continue
        miss = [k for k in keys if k not in got]
        extra = [k for k in got if k not in keys]
        print(f"[audit] {name}: cloud fields={len(got)} expected={len(keys)} missing={len(miss)} extra={len(extra)}")
        if miss:
            print("        MISSING:", miss[:60])
        if extra:
            print("        EXTRA  :", extra[:60])


if __name__ == "__main__":
    main()

"""Build corrected cloud blobs from device A's local save files.

- schema (which key holds which global): from the game's own save code
  (gml_Object_o_popup_option_Step_0.gml)
- values: from device A's INI files, decoded with the game's own load formulas
  (gameload_*.gml, including the +K obfuscation offsets)
- fields with no local source (e.g. save_land.ini absent, or cloud-only fields)
  keep the value currently stored in the cloud.
"""
import json, os, re, sys, shutil, time

# 目录约定（都在 BASE 下）：dump_all/CodeEntries（反编译产物）、saveA/files（本地存档）、
# cloud_raw/values（云端现值，playfab_fetch.py 的输出）。BASE 取脚本所在目录，或用 argv[1] 指定。
BASE = os.path.abspath(sys.argv[1]) if len(sys.argv) > 1 else os.path.dirname(os.path.abspath(__file__))
CE = os.path.join(BASE, "dump_all/CodeEntries")
SAVE = os.path.join(BASE, "saveA/files")
CLOUD = os.path.join(BASE, "cloud_raw/values")

# ---------- 1. global -> (ini, section, key, op, offset) ----------
LOADERS = {
    "gml_GlobalScript_gameload_info.gml": "save_info.ini",
    "gml_GlobalScript_gameload_data.gml": "save_data.ini",
    "gml_GlobalScript_gameload_stat.gml": "save_stat.ini",
    "gml_GlobalScript_gameload_unit.gml": "save_unit.ini",
    "gml_GlobalScript_gameload_item.gml": "save_item.ini",
    "gml_GlobalScript_gameload_menu.gml": "save_menu.ini",
    "gml_GlobalScript_gameload_achievement.gml": "save_achievement.ini",
    "gml_GlobalScript_gameload_land.gml": "save_land.ini",
    "gml_GlobalScript_gameload_deck.gml": "save_deck.ini",
}
# global = [floor(] ini_read_real("sec","key",def) [)] [ -/+ K] ;
READ = re.compile(
    r'global\.([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:(floor|round|int64)\()?\s*'
    r'ini_read_(real|string)\("([^"]+)",\s*"([^"]+)",\s*([^)]*)\)\s*\)?\s*([-+]\s*[0-9.]+)?\s*;')
# variant where the offset sits inside the floor(): floor(ini_read_real(s,k,d) - 84)
READ_INNER = re.compile(
    r'global\.([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:(floor|round|int64)\()?\s*'
    r'ini_read_(real|string)\("([^"]+)",\s*"([^"]+)",\s*([^(]*?)\)\s*([-+]\s*[0-9.]+)\)\s*;')
gl_map, unsupported = {}, []
for fn, inif in LOADERS.items():
    p = os.path.join(CE, fn)
    if not os.path.exists(p):
        unsupported.append((fn, "missing code entry"))
        continue
    for line in open(p, encoding='utf-8', errors='replace'):
        line = line.strip()
        if 'ini_read_' not in line:
            continue
        m = READ_INNER.match(line) or READ.match(line)
        if not m:
            unsupported.append((fn, line[:110])); continue
        g, fnc, kind, sec, key, dflt, off = m.groups()
        off = float(off.replace(' ', '')) if off else 0.0
        gl_map[g] = (inif, sec, key, kind, fnc, off)

print(f"[i] globals mapped from local save: {len(gl_map)}; unparsed loader lines: {len(unsupported)}")
for u in unsupported[:8]:
    print("    unparsed:", u)

def read_ini(path):
    d, sec = {}, None
    if not os.path.exists(path):
        return d
    for line in open(path, encoding='utf-8', errors='replace'):
        line = line.strip()
        if not line:
            continue
        if line.startswith('['):
            sec = line.strip('[]'); continue
        if '=' in line:
            k, v = line.split('=', 1)
            v = v.strip().strip('"')
            try:
                d[(sec, k.strip())] = float(v)
            except ValueError:
                d[(sec, k.strip())] = v
    return d

INIS = {f: read_ini(os.path.join(SAVE, f)) for f in set(v[0] for v in gl_map.values())}
print("[i] local ini loaded:", {k: len(v) for k, v in INIS.items()})

def local_value(g):
    """in-game value of a global, as the game would have it after boot."""
    m = gl_map.get(g)
    if not m:
        return None
    inif, sec, key, kind, fnc, off = m
    raw = INIS.get(inif, {}).get((sec, key))
    if raw is None:
        return None
    if kind == 'string':
        return raw
    try:
        v = float(raw) + off
    except (TypeError, ValueError):
        return None
    if fnc:
        v = int(v // 1) if v >= 0 else -int((-v) // 1)
    if v == int(v):
        return int(v)
    return round(v, 6)

# ---------- 2. schema from the game's save code ----------
lines = open(os.path.join(CE, "gml_Object_o_popup_option_Step_0.gml"), encoding='utf-8', errors='replace').read().split('\n')
BOUNDS = {"userdata": (25, 1735), "unitdata": (1750, 2104), "etcdata": (2119, 2894)}
FIELD = re.compile(r'^\s*([A-Za-z_][A-Za-z0-9_]*):\s*(.+?),?\s*$')

def schema(a, b):
    out = []
    for ln in lines[a - 1:b]:
        m = FIELD.match(ln.rstrip())
        if m:
            out.append((m.group(1), m.group(2).strip()))
    return out

STATS = {}
built = {}
for blob, (a, b) in BOUNDS.items():
    cur = json.loads(open(os.path.join(CLOUD, blob + ".json"), encoding='utf-8').read())
    fields = schema(a, b)
    out, cnt = {}, {"from_local": 0, "kept_cloud": 0, "only_cloud": 0, "changed": 0, "same": 0}
    for key, expr in fields:
        if key == blob:                      # wrapper key of the outer object
            continue
        m = re.fullmatch(r'global\.([A-Za-z_][A-Za-z0-9_]*)', expr)
        val = local_value(m.group(1)) if m else None
        if val is not None:
            out[key] = val; cnt["from_local"] += 1
        elif key in cur:
            out[key] = cur[key]; cnt["kept_cloud"] += 1
        else:
            out[key] = 0; cnt["only_cloud"] += 1
        if key in cur:
            if out[key] == cur[key]:
                cnt["same"] += 1
            else:
                cnt["changed"] += 1
    extra = [k for k in cur if k not in out]
    built[blob] = out
    STATS[blob] = (cnt, len(fields), extra)
    print(f"\n== {blob}: fields={len(fields)} extra-in-cloud={len(extra)} {extra[:5]}")
    print("   ", cnt)

# ---------- 3. sanity: what the game will read back for the crash-critical fields ----------
u = built["userdata"]
print("\n[crash-critical fields in the rebuilt userdata]")
for k, what in [("a27", "p1_select (1=CROW 2=RAVEN)"), ("a7", "p1_gameturn (DAY)"), ("a28", "difficulty"),
                ("a29", "difficulty_style"), ("a3", "p1_gold"), ("a4", "p1_supplies"), ("a5", "p1_medicine"),
                ("a6", "p1_cash"), ("a1", "check_tutorial"), ("a2", "p1_gameclear"), ("a9", "newgame"),
                ("a24", "bgm_ok"), ("a25", "se_ok"), ("a26", "set_language"), ("a11", "p1_pioneer"),
                ("a13", "p1_gameturn2"), ("a15", "proposal_count"), ("a12", "event_boss_num"),
                ("a30", "ope_num"), ("a31", "option_battle_speed"), ("a32", "map_tap_now")]:
    print(f"   {k:>5} = {u.get(k)!r:>12}   {what}")
print("   h1..h10 =", [u.get('h%d' % i) for i in range(1, 11)])
print("   h41..h48 =", [u.get('h%d' % i) for i in range(41, 49)])

# crash simulation: every key the load code reads for check_event_sub2_* / mission_clear
ev = {k: u.get(k) for k in ('h1', 'h2', 'h3', 'h4', 'h5')}
print("   event flags ->", ev)
print("   mission_clear-ish fields present:", sum(1 for k in u if k.startswith('g')))

out_dir = os.path.join(BASE, "cloud_build")
os.makedirs(out_dir, exist_ok=True)
for blob, obj in built.items():
    raw = json.dumps(obj, separators=(',', ':'))
    open(os.path.join(out_dir, blob + ".json"), "w", encoding="utf-8").write(raw)
    print(f"\n[i] wrote cloud_build/{blob}.json  ({len(raw)} bytes, {len(obj)} fields)")
print("\n[i] current cloud sizes:", {b: os.path.getsize(os.path.join(CLOUD, b + ".json")) for b in built})

#!/usr/bin/env python
"""Push the rebuilt cloud blobs (cloud_build/*.json) to the user's PlayFab save.

Backs up the current values first, then Client/UpdateUserData, then re-reads the
cloud and verifies. Never prints the session ticket.
"""
import json, os, shutil, sys, time, urllib.request, urllib.error

TITLE = "142A3"
BASE = os.path.dirname(os.path.abspath(__file__))
CLOUD = os.path.join(BASE, "cloud_raw")
BUILD = os.path.join(BASE, "cloud_build")
BACKUP = os.path.join(BASE, "cloud_backup")


def post(path, body, ticket=None):
    req = urllib.request.Request(f"https://{TITLE}.playfabapi.com/{path}",
                                 data=json.dumps(body).encode(), method="POST")
    req.add_header("Content-Type", "application/json")
    if ticket:
        req.add_header("X-Authorization", ticket)
    try:
        with urllib.request.urlopen(req, timeout=90) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"raw": raw[:2000]}


def main():
    args = [a for a in sys.argv[1:] if not a.startswith("-")]
    dry = "--dry" in sys.argv
    pid = args[0].strip() if args else open(os.path.join(BASE, "player_id.txt")).read().strip()

    st, resp = post("Client/LoginWithCustomID",
                    {"TitleId": TITLE, "CustomId": pid, "CreateAccount": False})
    if resp.get("code") != 200:
        print("[x] login failed:", resp.get("errorCode"), resp.get("errorMessage")); return 1
    ticket = resp["data"]["SessionTicket"]
    print("[+] logged in, PlayFabId:", resp["data"].get("PlayFabId"))

    # ---- backup current cloud values ----
    st, resp = post("Client/GetUserData", {}, ticket)
    if resp.get("code") != 200:
        print("[x] GetUserData failed:", resp.get("errorCode")); return 1
    cur = resp["data"]["Data"]
    ts = time.strftime("%Y%m%d-%H%M%S")
    os.makedirs(BACKUP, exist_ok=True)
    bpath = os.path.join(BACKUP, f"cloud_backup_{ts}.json")
    json.dump({k: {"Value": v.get("Value"), "LastUpdated": v.get("LastUpdated"),
                   "Permission": v.get("Permission")} for k, v in cur.items()},
              open(bpath, "w", encoding="utf-8"), indent=1)
    print(f"[+] backed up current cloud data -> {bpath}")
    for k, v in sorted(cur.items()):
        print(f"    {k}: {len(v.get('Value') or '')} bytes, LastUpdated {v.get('LastUpdated')}, {v.get('Permission')}")

    payload = {}
    for blob in ("userdata", "unitdata", "etcdata"):
        payload[blob] = open(os.path.join(BUILD, blob + ".json"), encoding="utf-8").read()
    print("[i] upload sizes:", {k: len(v) for k, v in payload.items()})
    if dry:
        print("[dry-run] not uploading"); return 0

    st, resp = post("Client/UpdateUserData", {"Data": payload}, ticket)
    print("[upload] http", st, "->", resp.get("code"), resp.get("status"),
          resp.get("errorCode") or "", resp.get("errorMessage") or "")
    if resp.get("code") != 200:
        print("[x] upload FAILED; backup:", bpath)
        return 1
    print("[+] UpdateUserData accepted:", resp.get("data"))

    # ---- verify by reading back ----
    st, resp = post("Client/GetUserData", {}, ticket)
    data = resp.get("data", {}).get("Data", {}) or {}
    print("[i] read-back raw:", json.dumps(resp)[:400])
    ok = True
    for k in sorted(payload):
        if k not in data:
            print(f"    {k}: MISSING after upload"); ok = False; continue
        val = data[k].get("Value") or ""
        same = val == payload[k]
        ok &= same
        print(f"    {k}: {len(val)} bytes, LastUpdated {data[k].get('LastUpdated')}, "
              f"{data[k].get('Permission')}, matches_upload={same}")
    if ok:
        u = json.loads(open(os.path.join(BUILD, "userdata.json"), encoding="utf-8").read())
        print("[verify] p1_select(a27) =", u.get("a27"), "| DAY(a7) =", u.get("a7"),
              "| difficulty(a28) =", u.get("a28"), "| check_event_sub2_1(h1) =", u.get("h1"))
        print("[+] cloud save replaced and verified.")
    else:
        print("[!] verify mismatch")
    return 0


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

#!/usr/bin/env python
"""Restore the PlayFab cloud save from a backup file made by playfab_push.py.

Usage: python playfab_restore.py cloud_backup/cloud_backup_YYYYmmdd-HHMMSS.json
"""
import json, os, sys, urllib.request, urllib.error

TITLE = "142A3"
BASE = os.path.dirname(os.path.abspath(__file__))


def post(path, body, ticket=None):
    req = urllib.request.Request(f"https://{TITLE}.playfabapi.com/{path}",
                                 data=json.dumps(body).encode(), method="POST")
    req.add_header("Content-Type", "application/json")
    if ticket:
        req.add_header("X-Authorization", ticket)
    try:
        with urllib.request.urlopen(req, timeout=90) as r:
            return r.status, json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        raw = e.read().decode()
        try:
            return e.code, json.loads(raw)
        except Exception:
            return e.code, {"raw": raw[:2000]}


def main():
    path = sys.argv[1]
    pid = open(os.path.join(BASE, "player_id.txt")).read().strip()
    st, resp = post("Client/LoginWithCustomID", {"TitleId": TITLE, "CustomId": pid, "CreateAccount": False})
    if resp.get("code") != 200:
        print("[x] login failed:", resp.get("errorCode"), resp.get("errorMessage")); return 1
    ticket = resp["data"]["SessionTicket"]
    backup = json.load(open(path, encoding="utf-8"))
    payload = {k: (v.get("Value") if isinstance(v, dict) else v) for k, v in backup.items()}
    print("[i] restoring", {k: len(v or "") for k, v in payload.items()})
    st, resp = post("Client/UpdateUserData", {"Data": payload}, ticket)
    print("[restore] ->", resp.get("code"), resp.get("status"), resp.get("errorCode") or "",
          resp.get("errorMessage") or "", resp.get("data"))
    st, resp = post("Client/GetUserData", {}, ticket)
    data = resp.get("data", {}).get("Data", {}) or {}
    for k in sorted(payload):
        got = (data.get(k) or {}).get("Value") or ""
        print(f"    {k}: {len(got)} bytes, matches_backup={got == payload[k]}")
    return 0


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