| #!/usr/bin/env python3 | | """ | | Minion Rush savegame encoder, version 1.2.0. | | | | Rebuilds a Windows Store Minion Rush savegame from JSON produced by | | minion_save_decoder_v2_modified.py. | | | | A template savegame is required because the decoded JSON intentionally does not | | contain every opaque RedundantStream marker byte. The encoder preserves the | | physical wrapper and replaces each valid redundant payload copy in-place. | | | | Typical use: | | python minion_save_encoder_v1.py savegame.json --template savegame --out savegame.new | | | | The top-level decoded_values object is applied to the Player and MapMgr blobs. | | Optionally, the same banana/token values can also be written to a raw | | connectivity_trackers file using --connectivity-template/--connectivity-out. | | | | Important: version-0x1C Player data contains an on-disk wallet shadow near the | | end of the serialized blob. When its valid flag is 1, the deserializer uses the | | shadow values to overwrite the ordinary banana/token fields. This encoder patches | | both representations together. | | | | Fields: | | banana_count -> Player + 0x10 (u32 little-endian) | | token_count -> Player + 0x0C (u32 little-endian) | | current_jelly_lab_level -> MapMgr + 0x08 (u32 little-endian) | | """ | | | | from future import annotations | | | | import argparse | | import base64 | | import copy | | import json | | import math | | import struct | | import sys | | import zlib | | from pathlib import Path | | from typing import Any | | | | HEADER_MARKER_SIZE = 0xB0 | | HEADER_SIZE = HEADER_MARKER_SIZE + 8 | | PAYLOAD_MARKER_SIZE = 0x90 | | SAVE_MAGIC = 0xED | | DEFAULT_KEY_SOURCE = b"ERROR: invalid stream" | | MAX_U32 = 0xFFFFFFFF | | VERSION = "1.2.0" | | | | | | class EncodeError(Exception): | | pass | | | | | | def derive_xtea_key(source: bytes) -> bytes: | | key = bytearray(16) | | for i, byte in enumerate(source): | | key[i & 0x0F] ^= byte | | return bytes(key) | | | | | | def xtea_encrypt_block(block: bytes, key: bytes) -> bytes: | | if len(block) != 8: | | raise ValueError("XTEA block must be exactly 8 bytes") | | if len(key) != 16: | | raise ValueError("XTEA key must be exactly 16 bytes") | | | | v0, v1 = struct.unpack("<2I", block) | | k = struct.unpack("<4I", key) | | delta = 0x9E3779B9 | | total = 0 | | | | for _ in range(32): | | v0 = ( | | v0 | |
- ( | | (((v1 << 4) ^ (v1 >> 5)) + v1) | | ^ ((total + k[total & 3]) & MAX_U32) | | ) | | ) & MAX_U32 | | total = (total + delta) & MAX_U32 | | v1 = ( | | v1 | |
- ( | | (((v0 << 4) ^ (v0 >> 5)) + v0) | | ^ ((total + k[(total >> 11) & 3]) & MAX_U32) | | ) | | ) & MAX_U32 | | | | return struct.pack("<2I", v0, v1) | | | | | | def xtea_encrypt(data: bytes, key: bytes) -> bytes: | | if len(data) % 8: | | raise EncodeError("XTEA plaintext must be padded to a multiple of 8 bytes") | | return b"".join( | | xtea_encrypt_block(data[offset:offset + 8], key) | | for offset in range(0, len(data), 8) | | ) | | | | | | def parse_key(args: argparse.Namespace) -> bytes: | | supplied = sum( | | value is not None | | for value in (args.key_source, args.key_source_hex, args.xtea_key_hex) | | ) | | if supplied > 1: | | raise EncodeError( | | "use only one of --key-source, --key-source-hex, or --xtea-key-hex" | | ) | | | | if args.xtea_key_hex is not None: | | try: | | key = bytes.fromhex(args.xtea_key_hex) | | except ValueError as exc: | | raise EncodeError(f"invalid --xtea-key-hex: {exc}") from exc | | if len(key) != 16: | | raise EncodeError("--xtea-key-hex must decode to exactly 16 bytes") | | return key | | | | if args.key_source_hex is not None: | | try: | | source = bytes.fromhex(args.key_source_hex) | | except ValueError as exc: | | raise EncodeError(f"invalid --key-source-hex: {exc}") from exc | | return derive_xtea_key(source) | | | | if args.key_source is not None: | | return derive_xtea_key(args.key_source.encode(args.key_encoding)) | | | | return derive_xtea_key(DEFAULT_KEY_SOURCE) | | | | | | def checked_u32(value: Any, name: str) -> int: | | if isinstance(value, bool) or not isinstance(value, int): | | raise EncodeError(f"{name} must be an integer") | | if not 0 <= value <= MAX_U32: | | raise EncodeError(f"{name} must be in the range 0..{MAX_U32}") | | return value | | | | | | | | def rol32(value: int, count: int) -> int: | | count &= 31 | | value &= MAX_U32 | | return ((value << count) | (value >> ((32 - count) & 31))) & MAX_U32 | | | | | | def wallet_shadow_encode(value: int) -> int: | | return rol32(checked_u32(value, "wallet shadow value"), 10) ^ 0x48607922 | | | | | | def patch_wallet_shadow( | | blob: bytearray, | | old_tokens: int, | | old_bananas: int, | | new_tokens: int, | | new_bananas: int, | | ) -> dict[str, Any]: | | """Patch the serialized wallet shadow used by Player format 0x1C. | | | | Serialized layout is contiguous: | | uint8 valid_flag == 1 | | uint32 token_shadow | | uint32 banana_shadow | | | | shadow(value) = ROL32(value, 10) XOR 0x48607922. | | """ | | old_seq = ( | | b"\x01" | |
- struct.pack("<I", wallet_shadow_encode(old_tokens)) | |
- struct.pack("<I", wallet_shadow_encode(old_bananas)) | | ) | | new_seq = ( | | b"\x01" | |
- struct.pack("<I", wallet_shadow_encode(new_tokens)) | |
- struct.pack("<I", wallet_shadow_encode(new_bananas)) | | ) | | matches = [] | | start = 0 | | while True: | | pos = blob.find(old_seq, start) | | if pos < 0: | | break | | matches.append(pos) | | start = pos + 1 | | | | if len(matches) != 1: | | raise EncodeError( | | "could not identify a unique version-0x1C wallet shadow: " | | f"expected one sequence for tokens={old_tokens}, bananas={old_bananas}, " | | f"found {len(matches)}" | | ) | | | | offset = matches[0] | | blob[offset:offset + len(old_seq)] = new_seq | | return { | | "offset": offset, | | "old_token_shadow": f"0x{wallet_shadow_encode(old_tokens):08X}", | | "old_banana_shadow": f"0x{wallet_shadow_encode(old_bananas):08X}", | | "new_token_shadow": f"0x{wallet_shadow_encode(new_tokens):08X}", | | "new_banana_shadow": f"0x{wallet_shadow_encode(new_bananas):08X}", | | } | | | | def decode_blob(record: dict[str, Any], name: str) -> bytearray: | | if record.get("kind") != "blob" or "data_base64" not in record: | | raise EncodeError(f"record {name!r} is not a Base64 blob") | | try: | | return bytearray(base64.b64decode(record["data_base64"], validate=True)) | | except (ValueError, TypeError) as exc: | | raise EncodeError(f"record {name!r} has invalid data_base64") from exc | | | | | | def patch_u32(blob: bytearray, offset: int, value: Any, field: str) -> None: | | value = checked_u32(value, field) | | if offset + 4 > len(blob): | | raise EncodeError( | | f"cannot write {field}: blob is {len(blob)} bytes, offset is 0x{offset:X}" | | ) | | struct.pack_into("<I", blob, offset, value) | | | | | | def apply_decoded_values(document: dict[str, Any]) -> dict[str, Any]: | | records = document.get("records") | | if not isinstance(records, dict): | | raise EncodeError("JSON root must contain a 'records' object") | | | | report: dict[str, Any] = {} | | top = document.get("decoded_values") | | if top is None: | | top = {} | | if not isinstance(top, dict): | | raise EncodeError("decoded_values must be an object") | | | | player = records.get("Player") | | if isinstance(player, dict): | | values: dict[str, Any] = {} | | if isinstance(player.get("decoded"), dict): | | values.update(player["decoded"]) | | values.update({k: top[k] for k in ("token_count", "banana_count") if k in top}) | | | | if values: | | blob = decode_blob(player, "Player") | | old_tokens = struct.unpack_from("<I", blob, 0x0C)[0] | | old_bananas = struct.unpack_from("<I", blob, 0x10)[0] | | new_tokens = checked_u32(values.get("token_count", old_tokens), "token_count") | | new_bananas = checked_u32(values.get("banana_count", old_bananas), "banana_count") | | | | shadow_report = patch_wallet_shadow( | | blob, old_tokens, old_bananas, new_tokens, new_bananas | | ) | | patch_u32(blob, 0x0C, new_tokens, "token_count") | | patch_u32(blob, 0x10, new_bananas, "banana_count") | | | | player["data_base64"] = base64.b64encode(blob).decode("ascii") | | player["length"] = len(blob) | | player["_aux"] = len(blob) | | report["Player"] = { | | "old_token_count": old_tokens, | | "old_banana_count": old_bananas, | | "new_token_count": new_tokens, | | "new_banana_count": new_bananas, | | "wallet_shadow": shadow_report, | | } | | | | map_mgr = records.get("MapMgr") | | if isinstance(map_mgr, dict): | | values = {} | | if isinstance(map_mgr.get("decoded"), dict): | | values.update(map_mgr["decoded"]) | | if "current_jelly_lab_level" in top: | | values["current_jelly_lab_level"] = top["current_jelly_lab_level"] | | | | if "current_jelly_lab_level" in values: | | blob = decode_blob(map_mgr, "MapMgr") | | patch_u32( | | blob, | | 0x08, | | values["current_jelly_lab_level"], | | "current_jelly_lab_level", | | ) | | map_mgr["data_base64"] = base64.b64encode(blob).decode("ascii") | | map_mgr["length"] = len(blob) | | map_mgr["_aux"] = len(blob) | | report["MapMgr"] = { | | "current_jelly_lab_level": values["current_jelly_lab_level"] | | } | | | | return report | | | | def encode_record_string(value: str) -> bytes: | | if not isinstance(value, str): | | raise EncodeError("record name/string value must be a string") | | raw = value.encode("utf-8") | | if len(raw) > 0xFFFF: | | raise EncodeError("record name/string exceeds 65535 encoded bytes") | | return struct.pack("<H", len(raw)) + raw | | | | | | def parse_nonfinite(value: Any, field: str) -> float: | | if isinstance(value, (int, float)) and not isinstance(value, bool): | | return float(value) | | if isinstance(value, str): | | mapping = { | | "nan": math.nan, | | "inf": math.inf, | | "+inf": math.inf, | | "-inf": -math.inf, | | } | | parsed = mapping.get(value.lower()) | | if parsed is not None: | | return parsed | | raise EncodeError(f"{field} must be a number, 'nan', 'inf', or '-inf'") | | | | | | def encode_record_value(record: dict[str, Any]) -> bytes: | | if not isinstance(record, dict): | | raise EncodeError("each record value must be an object") | | try: | | type_id = int(record["_type"]) | | except (KeyError, TypeError, ValueError) as exc: | | raise EncodeError("record is missing a valid _type") from exc | | | | body: bytes | | aux: int | | | | if type_id == 0: | | encoded = record.get("data_base64", "") | | try: | | body = base64.b64decode(encoded, validate=True) if encoded else b"" | | except (ValueError, TypeError) as exc: | | raise EncodeError("type-0 record has invalid data_base64") from exc | | aux = len(body) | | | | elif type_id == 1: | | if "raw" in record: | | raw = checked_u32(record["raw"], "type-1 raw") | | else: | | raw = 1 if bool(record.get("value")) else 0 | | body = struct.pack("<I", raw) | | aux = checked_u32(record.get("_aux", 0), "type-1 _aux") | | | | elif type_id == 2: | | raw_hex = record.get("raw_hex") | | if raw_hex is not None: | | try: | | body = bytes.fromhex(raw_hex) | | except ValueError as exc: | | raise EncodeError("type-2 raw_hex is invalid") from exc | | if len(body) != 8: | | raise EncodeError("type-2 raw_hex must contain 8 bytes") | | elif "unsigned" in record: | | body = struct.pack("<Q", int(record["unsigned"])) | | elif "signed" in record: | | body = struct.pack("<q", int(record["signed"])) | | else: | | raise EncodeError("type-2 record needs raw_hex, unsigned, or signed") | | aux = checked_u32(record.get("_aux", 0), "type-2 _aux") | | | | elif type_id == 3: | | raw_hex = record.get("raw_hex") | | if raw_hex is not None: | | try: | | body = bytes.fromhex(raw_hex) | | except ValueError as exc: | | raise EncodeError("type-3 raw_hex is invalid") from exc | | if len(body) != 4: | | raise EncodeError("type-3 raw_hex must contain 4 bytes") | | elif "unsigned" in record: | | body = struct.pack("<I", checked_u32(record["unsigned"], "unsigned")) | | elif "signed" in record: | | body = struct.pack("<i", int(record["signed"])) | | else: | | raise EncodeError("type-3 record needs raw_hex, unsigned, or signed") | | aux = checked_u32(record.get("_aux", 0), "type-3 _aux") | | | | elif type_id == 4: | | raw_hex = record.get("raw_hex") | | if raw_hex is not None: | | try: | | body = bytes.fromhex(raw_hex) | | except ValueError as exc: | | raise EncodeError("type-4 raw_hex is invalid") from exc | | if len(body) != 4: | | raise EncodeError("type-4 raw_hex must contain 4 bytes") | | else: | | body = struct.pack("<f", parse_nonfinite(record.get("value"), "float32")) | | aux = checked_u32(record.get("_aux", 0), "type-4 _aux") | | | | elif type_id == 5: | | raw_hex = record.get("raw_hex") | | if raw_hex is not None: | | try: | | body = bytes.fromhex(raw_hex) | | except ValueError as exc: | | raise EncodeError("type-5 raw_hex is invalid") from exc | | if len(body) != 8: | | raise EncodeError("type-5 raw_hex must contain 8 bytes") | | else: | | body = struct.pack("<d", parse_nonfinite(record.get("value"), "float64")) | | aux = checked_u32(record.get("_aux", 0), "type-5 _aux") | | | | elif type_id == 6: | | body = encode_record_string(record.get("value", "")) | | aux = checked_u32(record.get("_aux", 0), "type-6 _aux") | | | | elif type_id == 7: | | body = bytes(decode_blob(record, "<blob>")) | | aux = len(body) | | | | elif type_id == 8: | | nested = record.get("value") | | if not isinstance(nested, dict): | | raise EncodeError("type-8 record is missing nested 'value' RecordDB") | | body = encode_record_db(nested) | | aux = checked_u32(record.get("_aux", 0), "type-8 _aux") | | | | else: | | raise EncodeError(f"unsupported RecordDB type {type_id}") | | | | return struct.pack("<BI", type_id, aux) + body | | | | | | def ordered_entries(database: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]: | | records = database.get("records") | | if not isinstance(records, dict): | | raise EncodeError("RecordDB object is missing a 'records' object") | | | | entries: list[tuple[str, dict[str, Any]]] = [] | | duplicate_names: set[str] = set() | | duplicates = database.get("_duplicates", {}) | | | | if isinstance(duplicates, dict): | | for name, group in duplicates.items(): | | if not isinstance(group, list): | | raise EncodeError(f"duplicate group {name!r} must be a list") | | duplicate_names.add(name) | | for record in group: | | if not isinstance(record, dict): | | raise EncodeError(f"duplicate record {name!r} must be an object") | | entries.append((name, record)) | | | | for name, record in records.items(): | | if not isinstance(record, dict): | | raise EncodeError(f"record {name!r} must be an object") | | | | if name not in duplicate_names: | | entries.append((name, record)) | | | | def sort_key(item: tuple[str, dict[str, Any]]) -> tuple[int, str]: | | index = item[1].get("_record_index") | | return (index if isinstance(index, int) else 1 << 60, item[0]) | | | | entries.sort(key=sort_key) | | return entries | | | | | | def encode_record_db(database: dict[str, Any]) -> bytes: | | entries = ordered_entries(database) | | out = bytearray(struct.pack("<I", len(entries))) | | for name, record in entries: | | out += encode_record_string(name) | | out += encode_record_value(record) | | return bytes(out) | | | | | | def build_logical_payload(recorddb: bytes, key: bytes) -> bytes: | | inner_crc = zlib.crc32(recorddb) & MAX_U32 | | plaintext = struct.pack("<I", inner_crc) + recorddb | | padded_length = (len(plaintext) + 7) & ~7 | | padded = plaintext + b"\x00" * (padded_length - len(plaintext)) | | ciphertext = xtea_encrypt(padded, key) | | section_length = 4 + len(ciphertext) | | return ( | | struct.pack("<BI", SAVE_MAGIC, 1) | |
- struct.pack("<II", section_length, len(plaintext)) | |
- ciphertext | | ) | | | | | | def find_all(data: bytes, pattern: bytes) -> list[int]: | | positions: list[int] = [] | | start = 0 | | while True: | | pos = data.find(pattern, start) | | if pos < 0: | | return positions | | positions.append(pos) | | start = pos + 1 | | | | | | def locate_template_payloads(data: bytes) -> tuple[int, int, list[int], list[int]]: | | if len(data) < 2 * HEADER_SIZE + PAYLOAD_MARKER_SIZE: | | raise EncodeError("template is too small to contain a RedundantStream wrapper") | | | | header_marker = data[:HEADER_MARKER_SIZE] | | stored_crc, payload_len = struct.unpack_from("<II", data, HEADER_MARKER_SIZE) | | header_positions = find_all(data, header_marker) | | header_set = set(header_positions) | | | | candidate_starts: list[int] = [] | | for pos in header_positions: | | if pos - HEADER_SIZE in header_set and pos + HEADER_SIZE not in header_set: | | candidate_starts.append(pos + HEADER_SIZE + PAYLOAD_MARKER_SIZE) | | for pos in header_positions: | | candidate_starts.append(pos + HEADER_SIZE + PAYLOAD_MARKER_SIZE) | | | | valid_starts: list[int] = [] | | for start in dict.fromkeys(candidate_starts): | | if start + payload_len <= len(data): | | payload = data[start:start + payload_len] | | if zlib.crc32(payload) & MAX_U32 == stored_crc: | | valid_starts.append(start) | | | | if not valid_starts: | | raise EncodeError("could not locate any CRC-valid payload copies in template") | | | | valid_headers: list[int] = [] | | for pos in header_positions: | | if pos + HEADER_SIZE <= len(data): | | crc, length = struct.unpack_from("<II", data, pos + HEADER_MARKER_SIZE) | | if crc == stored_crc and length == payload_len: | | valid_headers.append(pos) | | | | return stored_crc, payload_len, sorted(set(valid_starts)), valid_headers | | | | | | def rewrite_template(template: bytes, logical_payload: bytes) -> tuple[bytes, int, int]: | | _, old_length, payload_starts, header_positions = locate_template_payloads(template) | | if len(logical_payload) != old_length: | | raise EncodeError( | | "encoded logical payload length changed from " | | f"{old_length} to {len(logical_payload)} bytes. This encoder currently " | | "supports in-place, same-length edits only. Editing decoded counters works; " | | "adding/removing records or changing string/blob lengths does not." | | ) | | | | result = bytearray(template) | | new_crc = zlib.crc32(logical_payload) & MAX_U32 | | | | for start in payload_starts: | | result[start:start + old_length] = logical_payload | | | | for pos in header_positions: | | struct.pack_into("<II", result, pos + HEADER_MARKER_SIZE, new_crc, old_length) | | | | return bytes(result), len(payload_starts), len(header_positions) | | | | | | | | def parse_connectivity_trackers(data: bytes) -> tuple[int, int, list[dict[str, Any]]]: | | """Parse the observed compact connectivity_trackers container.""" | | if len(data) < 13: | | raise EncodeError("connectivity_trackers template is too small") | | declared_size, version, stored_crc = struct.unpack_from("<IBI", data, 0) | | body = data[9:] | | actual_crc = zlib.crc32(body) & MAX_U32 | | if actual_crc != stored_crc: | | raise EncodeError( | | "connectivity_trackers CRC mismatch: " | | f"stored={stored_crc:08x}, calculated={actual_crc:08x}" | | ) | | count = struct.unpack_from("<I", body, 0)[0] | | offset = 4 | | records: list[dict[str, Any]] = [] | | for index in range(count): | | if offset + 2 > len(body): | | raise EncodeError(f"truncated connectivity record {index}") | | name_len = struct.unpack_from("<H", body, offset)[0] | | offset += 2 | | if offset + name_len + 5 > len(body): | | raise EncodeError(f"truncated connectivity record {index}") | | try: | | name = body[offset:offset + name_len].decode("utf-8") | | except UnicodeDecodeError as exc: | | raise EncodeError(f"invalid connectivity record name at index {index}") from exc | | offset += name_len | | type_code, size = struct.unpack_from("<BI", body, offset) | | offset += 5 | | if offset + size > len(body): | | raise EncodeError(f"truncated connectivity value {name!r}") | | raw = body[offset:offset + size] | | offset += size | | records.append({"name": name, "type_code": type_code, "raw": raw}) | | if offset != len(body): | | raise EncodeError( | | f"connectivity_trackers has {len(body) - offset} unexpected trailing bytes" | | ) | | bias = declared_size - len(data) | | return version, bias, records | | | | | | def encode_connectivity_value(type_code: int, value: int, name: str) -> bytes: | | value = checked_u32(value, name) | | if type_code != 3: | | raise EncodeError( | | f"connectivity field {name!r} has type {type_code}; expected uint32 type 3" | | ) | | return struct.pack("<I", value) | | | | | | def patch_connectivity_trackers( | | template: bytes, | | banana_count: int | None, | | token_count: int | None, | | ) -> tuple[bytes, list[str]]: | | version, bias, records = parse_connectivity_trackers(template) | | patches = {"bananas": banana_count, "tokens": token_count} | | found: set[str] = set() | | body = bytearray(struct.pack("<I", len(records))) | | for rec in records: | | name = rec["name"] | | raw = rec["raw"] | | if name in patches and patches[name] is not None: | | raw = encode_connectivity_value(rec["type_code"], patches[name], name) | | found.add(name) | | name_raw = name.encode("utf-8") | | body += struct.pack("<H", len(name_raw)) + name_raw | | body += struct.pack("<BI", rec["type_code"], len(raw)) + raw | | requested = {name for name, value in patches.items() if value is not None} | | missing = requested - found | | if missing: | | raise EncodeError( | | "connectivity_trackers is missing requested fields: " + ", ".join(sorted(missing)) | | ) | | file_size = 9 + len(body) | | declared_size = file_size + bias | | crc = zlib.crc32(body) & MAX_U32 | | return struct.pack("<IBI", declared_size, version, crc) + body, sorted(found) | | | | | | def resolve_cli_values(document: dict[str, Any], args: argparse.Namespace) -> tuple[int | None, int | None]: | | top = document.setdefault("decoded_values", {}) | | if not isinstance(top, dict): | | raise EncodeError("decoded_values must be an object") | | if args.bananas is not None: | | top["banana_count"] = checked_u32(args.bananas, "--bananas") | | if args.tokens is not None: | | top["token_count"] = checked_u32(args.tokens, "--tokens") | | bananas = top.get("banana_count") | | tokens = top.get("token_count") | | if bananas is not None: | | bananas = checked_u32(bananas, "banana_count") | | if tokens is not None: | | tokens = checked_u32(tokens, "token_count") | | return bananas, tokens | | | | def main() -> int: | | parser = argparse.ArgumentParser( | | description="Encode decoder JSON back into a Minion Rush savegame" | | ) | | parser.add_argument("json_file", type=Path, help="JSON produced by the decoder") | | parser.add_argument("--template", required=True, type=Path, help="original savegame") | | parser.add_argument("--out", required=True, type=Path, help="output savegame path") | | parser.add_argument("--bananas", type=int, help="override Player banana count") | | parser.add_argument("--tokens", type=int, help="override Player token count") | | parser.add_argument( | | "--connectivity-template", | | type=Path, | | help="optional raw connectivity_trackers file to patch with the same balances", | | ) | | parser.add_argument( | | "--connectivity-out", | | type=Path, | | help="output path for patched connectivity_trackers", | | ) | | parser.add_argument("--key-source") | | parser.add_argument("--key-source-hex") | | parser.add_argument("--xtea-key-hex") | | parser.add_argument("--key-encoding", default="utf-8") | | args = parser.parse_args() | | | | try: | | document = json.loads(args.json_file.read_text(encoding="utf-8")) | | if not isinstance(document, dict): | | raise EncodeError("JSON root must be an object") | | | | document = copy.deepcopy(document) | | bananas, tokens = resolve_cli_values(document, args) | | patch_report = apply_decoded_values(document) | | recorddb = encode_record_db(document) | | key = parse_key(args) | | logical_payload = build_logical_payload(recorddb, key) | | template = args.template.read_bytes() | | output, payload_copies, headers = rewrite_template(template, logical_payload) | | args.out.write_bytes(output) | | | | connectivity_result = None | | if args.connectivity_template is not None or args.connectivity_out is not None: | | if args.connectivity_template is None or args.connectivity_out is None: | | raise EncodeError( | | "--connectivity-template and --connectivity-out must be used together" | | ) | | if bananas is None and tokens is None: | | raise EncodeError( | | "no banana/token values are available for connectivity_trackers" | | ) | | connectivity_template = args.connectivity_template.read_bytes() | | connectivity_output, patched_fields = patch_connectivity_trackers( | | connectivity_template, bananas, tokens | | ) | | args.connectivity_out.write_bytes(connectivity_output) | | connectivity_result = { | | "template": str(args.connectivity_template), | | "output": str(args.connectivity_out), | | "patched_fields": patched_fields, | | "output_size": len(connectivity_output), | | "output_crc32": f"{zlib.crc32(connectivity_output) & MAX_U32:08x}", | | } | | | | result = { | | "encoder_version": VERSION, | | "json": str(args.json_file), | | "template": str(args.template), | | "output": str(args.out), | | "recorddb_length": len(recorddb), | | "logical_payload_length": len(logical_payload), | | "redundant_payload_copies_replaced": payload_copies, | | "header_checksums_updated": headers, | | "output_size": len(output), | | "output_crc32": f"{zlib.crc32(output) & MAX_U32:08x}", | | "requested_balances": { | | "banana_count": bananas, | | "token_count": tokens, | | }, | | "patch_report": patch_report, | | "connectivity_trackers": connectivity_result, | | "wallet_note": ( | | "Patched the ordinary Player currency fields and the version-0x1C " | | "serialized wallet shadow used to overwrite them during load." | | ), | | } | | print(json.dumps(result, indent=2)) | | return 0 | | | | except (OSError, json.JSONDecodeError, EncodeError, ValueError, struct.error) as exc: | | print(f"error: {exc}", file=sys.stderr) | | return 1 | | | | | | if name == "main": | | raise SystemExit(main()) |