# Minion Rush Windows Store Edition save data editor (AI generated) > Source: > Published: 2026-07-28 08:34:52+00:00 | #!/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(" 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(" 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(" 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(" 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("")) | | 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(" 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") | | # For duplicate names, _duplicates already contains the first occurrence | | # as well as every later occurrence. | | 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(" bytes: | | inner_crc = zlib.crc32(recorddb) & MAX_U32 | | plaintext = struct.pack(" 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(" 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(" 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(" len(body): | | raise EncodeError(f"truncated connectivity record {index}") | | name_len = struct.unpack_from(" 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(" 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(" 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(" 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()) |