Lesson 4: Robust Saves: Integrity, Versioning, Security

🧨 What Can Go Wrong With a Save

In Saving & Loading you wrote JSON saves with a dataclass schema, a temporary file plus os.replace, validation and a first migration table. That covers the everyday cases. A game that ships to thousands of players meets the rest:

☠️ Never Unpickle a Save

pickle looks like the easy way to save anything: pickle.dump(game_state, f) and done. The problem is how loading works. A pickle file doesn't just hold data; it holds instructions for rebuilding objects, including "call this function with these arguments". Python's documentation says it plainly: never unpickle data you don't trust, because it can run arbitrary code. Here is a harmless demonstration:

import pickle


class NotReallyASave:
    def __reduce__(self):
        # pickle asks the object how to rebuild it; this one answers
        # "call print with this message". It could just as well name any other function.
        return (print, ("This ran just because you loaded a file!",))


booby_trapped = pickle.dumps(NotReallyASave())   # these bytes could be a downloaded "save"
print(len(booby_trapped), "bytes that look harmless")
pickle.loads(booby_trapped)                      # loading = running the attacker's choice

Output:

80 bytes that look harmless
This ran just because you loaded a file!

Replace print with a function that deletes files or downloads a program, and a "save file" posted on a forum becomes an attack on everyone who loads it. Players share saves all the time. So the rule for this course: saves are data formats only (JSON here), and every loaded value is validated before the game uses it.

📦 A Binary Container: struct, Byte Order, CRC

The payload stays JSON, compressed with zlib. In front of it goes a fixed 14-byte header that answers four questions before the game trusts anything: Is this our file? Which version wrote it? Is it all there? Is it intact?

The format string starts with <, and that character matters. Without it, struct uses your machine's native byte order, sizes and alignment padding, so a save written on one kind of machine may not read correctly on another. < means little-endian with no padding, the same on every machine:

import struct

print(struct.calcsize("4sHII"))      # native: size and padding depend on the machine
print(struct.calcsize("<4sHII"))     # little-endian, no padding: 14 on every machine
print(struct.pack("<I", 1))          # b'\x01\x00\x00\x00'  lowest byte first
print(struct.pack(">I", 1))          # b'\x00\x00\x00\x01'  highest byte first

On my x86-64 Linux machine the native version is 16 bytes, because two padding bytes are added after the H so the next I lines up; the little-endian version is always 14. The last two lines show byte order itself: the same number 1 stored lowest-byte-first and highest-byte-first. Pick one and write it into the format.

Here is the whole container as a complete program. It saves, loads, and then feeds the loader three kinds of bad file:

import json
import struct
import zlib

MAGIC = b"RSAV"
SCHEMA_VERSION = 3
HEADER = struct.Struct("<4sHII")        # magic, schema version, payload length, CRC-32


class CorruptSave(Exception):
    pass


def encode(data, version=SCHEMA_VERSION):
    payload = zlib.compress(json.dumps(data, separators=(",", ":")).encode("utf-8"))
    return HEADER.pack(MAGIC, version, len(payload), zlib.crc32(payload)) + payload


def decode(blob):
    if len(blob) < HEADER.size:
        raise CorruptSave("file too short for a header")
    magic, version, length, crc = HEADER.unpack_from(blob)
    if magic != MAGIC:
        raise CorruptSave("not a save file (wrong magic number)")
    payload = blob[HEADER.size:]
    if len(payload) != length:
        raise CorruptSave(f"truncated: expected {length} bytes, found {len(payload)}")
    if zlib.crc32(payload) != crc:
        raise CorruptSave("checksum mismatch")
    return version, json.loads(zlib.decompress(payload).decode("utf-8"))


save = {"pos": [120.5, 300.0], "coins": 42, "items": ["map", "lantern"], "hp": 80, "max_hp": 100}
blob = encode(save)
print(len(blob), "bytes; header:", blob[:HEADER.size].hex(" "))
print("round trip:", decode(blob))

damaged = bytearray(blob)
damaged[-3] ^= 0xFF                     # one flipped byte, as from a bad disk sector
for name, bad in [("flipped byte", bytes(damaged)), ("cut in half", blob[:len(blob) // 2]),
                  ("a PNG file", b"\x89PNG\r\n\x1a\n" + bytes(20))]:
    try:
        decode(bad)
    except CorruptSave as exc:
        print(f"{name}: rejected ({exc})")

Output:

93 bytes; header: 52 53 41 56 03 00 4f 00 00 00 ac f2 6a d6
round trip: (3, {'pos': [120.5, 300.0], 'coins': 42, 'items': ['map', 'lantern'], 'hp': 80, 'max_hp': 100})
flipped byte: rejected (checksum mismatch)
cut in half: rejected (truncated: expected 79 bytes, found 32)
a PNG file: rejected (not a save file (wrong magic number))

Read the header bytes against the table: 52 53 41 56 is "RSAV" in ASCII, 03 00 is version 3 (low byte first), 4f 00 00 00 is a 79-byte payload, and the last four bytes are the CRC. When a player sends you a broken save, this is exactly how you start the investigation.

The checks run cheapest first, and each one produces a message that says what is wrong. A CRC-32 is designed to catch accidental damage such as flipped bits and truncation. It is not a security feature, as the next sections show.

🔁 A Migration Chain

Your game will change its save layout. Version 1 stored x and y separately and items as one comma-separated string; version 2 made a pos list and an item list; version 3 renamed gold to coins and added hit points. Instead of writing a converter for every pair of versions, write one small function per step and chain them:

def v1_to_v2(data):
    return {"pos": [data["x"], data["y"]], "gold": data["gold"],
            "items": [name for name in data["items"].split(",") if name]}


def v2_to_v3(data):
    new = dict(data)                    # never change the caller's dict
    new["coins"] = new.pop("gold")
    new.setdefault("max_hp", 100)
    new.setdefault("hp", new["max_hp"])
    return new


MIGRATIONS = {1: v1_to_v2, 2: v2_to_v3}     # version N -> function that makes N + 1


def migrate(data, version):
    if version > SCHEMA_VERSION:
        raise UnsupportedVersion(f"save is v{version}, this game reads up to v{SCHEMA_VERSION}")
    while version < SCHEMA_VERSION:
        step = MIGRATIONS.get(version)
        if step is None:
            raise UnsupportedVersion(f"no migration from v{version}")
        data = step(data)
        version += 1
    return data

The rules that keep a chain trustworthy:

• The version is an integer in the header, so "is this older or newer?" is a plain comparison, and the loader knows the version before it parses the payload.
• Each step does one version. A v1 save runs v1_to_v2 then v2_to_v3; you never write v1_to_v3.
• Steps return new dictionaries and never throw data away: v1's items survive into v3, with empty names dropped on purpose.
• Newer versions are refused, not guessed at. An older build can't know what a newer build's fields mean.
• Keep real old saves as test files, one per version, and test that each still loads after every change.
• Validate after migrating, because a migrated save must meet the same rules as a new one.

🔐 Checksums, Signatures and "Encryption"

A CRC catches accidents. It does nothing against a player who wants 99,999 coins: they edit the JSON, compute a new CRC with the same zlib.crc32 call, and write it into the header. Stronger tools raise the bar, but all of them face the same fact: on the player's own computer, everything your game knows, the player can find.

If you do want to discourage casual edits, a signature from the standard library is enough; no extra package is needed:

import hashlib
import hmac

GAME_KEY = b"change-me-per-game"            # ships inside the game: obfuscation, not security


def sign(payload):
    return hmac.new(GAME_KEY, payload, hashlib.sha256).digest()     # 32 bytes


def verify(payload, signature):
    return hmac.compare_digest(sign(payload), signature)            # constant-time compare

Be honest with yourself about what that buys. For a single-player game, many developers decide save editing is the player's business and use a CRC only for corruption. For anything competitive, the only real protection is that the server decides what the player owns.

🛟 Atomic Writes and Backups

The atomic write from Saving & Loading gets two upgrades. First, os.fsync asks the operating system to push the temp file's bytes to the disk before the rename, so a power cut right after the rename doesn't leave an empty file. Second, the previous save is kept as a backup instead of being overwritten:

def write_save(path, data):
    path = Path(path)
    tmp = path.with_name(path.name + ".tmp")
    with open(tmp, "wb") as f:
        f.write(encode(data))
        f.flush()
        os.fsync(f.fileno())                    # bytes on disk before we swap names
    if path.exists():
        os.replace(path, path.with_name(path.name + ".bak"))   # keep the last good save
    os.replace(tmp, path)


def load_save(path):
    path = Path(path)
    try:
        return decode(path.read_bytes()), "loaded"
    except (OSError, SaveError) as main_error:
        backup = path.with_name(path.name + ".bak")
        try:
            return decode(backup.read_bytes()), f"{main_error}; loaded the backup"
        except (OSError, SaveError):
            raise SaveError(f"{main_error}; no usable backup") from main_error

Now every failure has a path forward: the main save is damaged, so load the backup and tell the player honestly what happened. Only when both files fail does the game have to say the save is lost, and it can still say why.

🏋️ Practice Exercise: Bulletproof Save Slot

"""Bulletproof Save Slot: Advanced Lesson 4 practice exercise (solution).

Walk with the arrow keys and collect coins. F5 saves, F9 loads.
The number keys attack your own save file so you can watch it survive:
    1  flip one byte in the save      2  cut the save file in half
    3  replace it with an old v1 save  4  replace it with a save from "v99"
Every save is a small binary container: a little-endian header with a magic
number, schema version, payload length and CRC-32, then zlib-compressed JSON.
Close the window to quit.
"""
import json
import os
import struct
import zlib
from pathlib import Path

import pygame


WIDTH, HEIGHT = 800, 480
SPEED = 220                             # px/s
PLAYER_SIZE = 24
COIN_RADIUS = 8
SAVE_DIR = Path(__file__).parent / "saves"
BG_COLOR = (16, 22, 34)
TEXT_COLOR = (226, 232, 240)

# --- the save container --------------------------------------------------------
MAGIC = b"RSAV"
SCHEMA_VERSION = 3
HEADER = struct.Struct("<4sHII")        # magic, schema version, payload length, CRC-32


class SaveError(Exception):
    """Base class: the save could not be used."""


class CorruptSave(SaveError):
    """The bytes are damaged, truncated or not a save at all."""


class UnsupportedVersion(SaveError):
    """The save is from a version this build cannot read."""


def v1_to_v2(data):
    """v1 stored x and y separately and items as one comma-separated string."""
    return {"pos": [data["x"], data["y"]], "gold": data["gold"],
            "items": [name for name in data["items"].split(",") if name]}


def v2_to_v3(data):
    """v3 renamed gold to coins and added hit points."""
    new = dict(data)                    # never change the caller's dict
    new["coins"] = new.pop("gold")
    new.setdefault("max_hp", 100)
    new.setdefault("hp", new["max_hp"])
    return new


MIGRATIONS = {1: v1_to_v2, 2: v2_to_v3}     # version N -> function that makes N + 1


def migrate(data, version):
    """Upgrade data one step at a time until it matches SCHEMA_VERSION."""
    if version > SCHEMA_VERSION:
        raise UnsupportedVersion(f"save is v{version}, this game reads up to v{SCHEMA_VERSION}")
    while version < SCHEMA_VERSION:
        step = MIGRATIONS.get(version)
        if step is None:
            raise UnsupportedVersion(f"no migration from v{version}")
        data = step(data)
        version += 1
    return data


def validate(data):
    """Reject well-formed saves whose values would break the game."""
    try:
        x, y = data["pos"]
        ok = (isinstance(x, (int, float)) and isinstance(y, (int, float))
              and isinstance(data["coins"], int) and data["coins"] >= 0
              and isinstance(data["items"], list)
              and all(isinstance(name, str) for name in data["items"])
              and 0 <= data["hp"] <= data["max_hp"])
    except (KeyError, TypeError, ValueError):
        ok = False
    if not ok:
        raise CorruptSave("values out of range or missing")
    return data


def encode(data, version=SCHEMA_VERSION):
    """dict -> bytes: header + zlib-compressed JSON."""
    payload = zlib.compress(json.dumps(data, separators=(",", ":")).encode("utf-8"))
    return HEADER.pack(MAGIC, version, len(payload), zlib.crc32(payload)) + payload


def decode(blob):
    """bytes -> validated dict at the current schema version, or raise SaveError."""
    if len(blob) < HEADER.size:
        raise CorruptSave("file too short for a header")
    magic, version, length, crc = HEADER.unpack_from(blob)
    if magic != MAGIC:
        raise CorruptSave("not a save file (wrong magic number)")
    payload = blob[HEADER.size:]
    if len(payload) != length:
        raise CorruptSave(f"truncated: expected {length} bytes, found {len(payload)}")
    if zlib.crc32(payload) != crc:
        raise CorruptSave("checksum mismatch")
    try:
        data = json.loads(zlib.decompress(payload).decode("utf-8"))
    except (zlib.error, UnicodeDecodeError, ValueError) as exc:
        raise CorruptSave(f"payload unreadable: {exc}") from exc
    return validate(migrate(data, version))


def write_save(path, data):
    """Atomic save that keeps the previous good save as <name>.bak."""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    with open(tmp, "wb") as f:
        f.write(encode(data))
        f.flush()
        os.fsync(f.fileno())            # the bytes are on disk before we swap names
    if path.exists():
        os.replace(path, path.with_name(path.name + ".bak"))
    os.replace(tmp, path)


def load_save(path):
    """Return (data, message). Falls back to the .bak file if the main save fails."""
    path = Path(path)
    try:
        return decode(path.read_bytes()), "loaded"
    except (OSError, SaveError) as main_error:
        backup = path.with_name(path.name + ".bak")
        try:
            return decode(backup.read_bytes()), f"{main_error}; loaded the backup"
        except (OSError, SaveError):
            raise SaveError(f"{main_error}; no usable backup") from main_error


# --- attacks on your own save file ------------------------------------------------
def flip_byte(path):
    blob = bytearray(Path(path).read_bytes())
    blob[-3] ^= 0xFF                     # damage the payload, not the header
    Path(path).write_bytes(bytes(blob))


def truncate(path):
    blob = Path(path).read_bytes()
    Path(path).write_bytes(blob[:len(blob) // 2])


def write_old_v1(path):
    old = {"x": 120, "y": 360, "gold": 7, "items": "sword,,lantern"}
    Path(path).write_bytes(encode(old, version=1))


def write_future(path):
    Path(path).write_bytes(encode({"pos": [0, 0], "coins": 0, "items": [], "hp": 1, "max_hp": 1},
                                  version=99))


# --- the game ---------------------------------------------------------------------
def new_game():
    coins = [pygame.Vector2(150 + 90 * i, 252) for i in range(7)]
    return {"pos": pygame.Vector2(80, 240), "coins": 0, "items": ["map"], "hp": 100,
            "max_hp": 100}, coins


def to_save(state):
    return {"pos": [state["pos"].x, state["pos"].y], "coins": state["coins"],
            "items": list(state["items"]), "hp": state["hp"], "max_hp": state["max_hp"]}


def from_save(data):
    state = dict(data)
    state["pos"] = pygame.Vector2(data["pos"])
    return state


def main():
    pygame.init()
    screen = pygame.display.set_mode((WIDTH, HEIGHT))
    pygame.display.set_caption("Bulletproof Save Slot")
    clock = pygame.time.Clock()
    font = pygame.font.Font(None, 24)
    save_path = SAVE_DIR / "slot1.rsav"

    state, coins = new_game()
    held = {pygame.K_LEFT: 0, pygame.K_RIGHT: 0, pygame.K_UP: 0, pygame.K_DOWN: 0}
    log = ["F5 save  F9 load  1 flip byte  2 truncate  3 old v1  4 future v99"]

    def report(message):
        print(message)
        log.append(message)
        del log[:-6]

    running = True
    while running:
        dt = clock.tick(60) / 1000
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.KEYUP and event.key in held:
                held[event.key] = 0
            elif event.type == pygame.KEYDOWN:
                if event.key in held:
                    held[event.key] = 1
                elif event.key == pygame.K_F5:
                    write_save(save_path, to_save(state))
                    report(f"saved: {state['coins']} coins, {save_path.stat().st_size} bytes")
                elif event.key == pygame.K_F9:
                    try:
                        data, message = load_save(save_path)
                        state = from_save(data)
                        report(f"load: {message} ({state['coins']} coins)")
                    except SaveError as exc:
                        report(f"load failed: {exc}")
                elif event.key in (pygame.K_1, pygame.K_2, pygame.K_3, pygame.K_4):
                    attack = {pygame.K_1: flip_byte, pygame.K_2: truncate,
                              pygame.K_3: write_old_v1, pygame.K_4: write_future}[event.key]
                    try:
                        attack(save_path)
                        report(f"attack: {attack.__name__}")
                    except OSError:
                        report("attack: save first (F5)")

        direction = pygame.Vector2(held[pygame.K_RIGHT] - held[pygame.K_LEFT],
                                   held[pygame.K_DOWN] - held[pygame.K_UP])
        if direction.length_squared() > 0:
            state["pos"] += direction.normalize() * SPEED * dt
        state["pos"].x = max(0, min(WIDTH - PLAYER_SIZE, state["pos"].x))
        state["pos"].y = max(0, min(HEIGHT - PLAYER_SIZE, state["pos"].y))
        player = pygame.FRect(state["pos"], (PLAYER_SIZE, PLAYER_SIZE))
        for coin in coins[:]:
            if player.collidepoint(coin):
                coins.remove(coin)
                state["coins"] += 1

        screen.fill(BG_COLOR)
        for coin in coins:
            pygame.draw.circle(screen, (250, 204, 21), coin, COIN_RADIUS)
        pygame.draw.rect(screen, (74, 222, 128), player)
        status = f"coins {state['coins']}   hp {state['hp']}/{state['max_hp']}   items {state['items']}"
        screen.blit(font.render(status, True, TEXT_COLOR), (12, 10))
        for n, line in enumerate(log):
            screen.blit(font.render(line, True, (148, 163, 184)), (12, HEIGHT - 20 - 22 * (len(log) - n)))
        pygame.display.flip()

    pygame.quit()
    print(f"Finished with {state['coins']} coins.")


if __name__ == "__main__":
    main()

📓 Learning Journal

📝 Summary

A robust save format assumes the file might be damaged, old, new or hostile. You kept the payload as JSON, because pickle can run code from a file, and wrapped it in a 14-byte little-endian header with a magic number, a version, a length and a CRC-32, checked cheapest first. A chain of one-step migrations upgrades old saves without losing data, while newer ones are refused. An fsynced atomic write with a .bak gives every failure a way back, and you learned to be honest about the limits: checksums catch accidents, and nothing on the player's machine truly stops a determined editor.

🔭 Looking Ahead

That completes the foundations: an architecture that scales, the tools to measure it, and saves that survive the real world. The next module turns to AI, starting with Steering & Flocking, where characters move with forces and flocks emerge from three simple rules.

❓ Common Questions

🎯 Quick Quiz

🌟 Going Further

• Quarantine broken saves: when both the save and its backup fail, rename the save to .broken instead of leaving it to be overwritten, and show the reason on screen.
• Several backups: keep the last three saves (.bak1, .bak2, .bak3) and try them newest first.
• A v4: add a play_time field in seconds, write v3_to_v4, and add a test that loads your old v1, v2 and v3 saves.
• Sign it: add the HMAC from this lesson as a 32-byte field after the header, and decide what the game should say when the signature fails.
• Read the docs: pickle (read the red warning box at the top), struct (byte order, size and alignment), zlib and hmac.