simple-updater/generate_manifest.py

import os
import json
import gzip
import hashlib
import argparse

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

# ============================================================
# CONFIGURATION - loaded from config.json, with hardcoded fallbacks below
# for anyone running this script without a config.json present (e.g. it
# was copied out on its own). Edit config.json, not this file - see
# README/quickstart.py for the intended workflow.
# ============================================================

_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_CONFIG_PATH = os.path.join(_SCRIPT_DIR, "config.json")


def _load_config():
    if not os.path.exists(_CONFIG_PATH):
        return {}
    try:
        with open(_CONFIG_PATH, "r", encoding="utf-8") as f:
            return json.load(f)
    except (OSError, json.JSONDecodeError):
        return {}


def _read_private_signing_key(cfg):
    """
    The private signing key lives in a standalone .txt file next to the
    scripts (see generate_key.py's write_private_key_file), NOT inside
    config.json - config.json only stores which filename to look for
    (signing_keys.private_key_file), so the secret itself never ends up
    in a file that's likely to get committed, zipped, or shared alongside
    the rest of the project's settings.

    Returns the hex key string, or None if the file doesn't exist or has
    no usable key in it (signing is then disabled - see SIGNING_PRIVATE_KEY
    below and the "unsigned manifest" warning where it's used).
    """
    filename = cfg.get("signing_keys", {}).get("private_key_file", "SIGNING_PRIVATE_KEY.txt")
    key_path = os.path.join(_SCRIPT_DIR, filename)
    if not os.path.exists(key_path):
        return None
    try:
        with open(key_path, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line.startswith("SIGNING_PRIVATE_KEY="):
                    return line.split("=", 1)[1].strip() or None
    except OSError:
        return None
    return None


_CFG = _load_config()
_manifest_cfg = _CFG.get("manifest", {})
_ignore_cfg = _CFG.get("ignore", {})

MANIFEST_FILENAME = _manifest_cfg.get("manifest_filename", "manifest.json")

# Suffix used for the precompressed copy of each file (e.g. "texture.dds"
# -> "texture.dds.gz").
COMPRESSED_SUFFIX = _manifest_cfg.get("compressed_suffix", ".gz")

# Where to put those precompressed copies, relative to the client folder.
# Empty string (default "") keeps the old behavior: "texture.dds.gz" sits
# right next to "texture.dds". Setting this to e.g. "_gz" moves ALL
# compressed copies into client_dir/_gz/ instead, mirroring the original
# folder structure underneath it (client_dir/_gz/system/game.exe.gz for
# client_dir/system/game.exe) - keeps the client folder itself free of
# .gz clutter. This only affects where generate_manifest.py writes files
# and where server.py looks for them (server.py falls back to this
# subfolder when a requested .gz isn't found at the direct path) -
# updater.py is completely unaware of this and always requests
# "<rel_path><compressed_suffix>" at what looks like the direct location.
COMPRESSED_DIR = _manifest_cfg.get("compressed_dir", "") or ""

# gzip compression level, 1 (fastest, larger) to 9 (slowest, smaller).
COMPRESSION_LEVEL = _manifest_cfg.get("compression_level", 6)

# Ed25519 PRIVATE key used to sign manifest.json, hex-encoded (32 raw bytes
# = 64 hex chars). Generate a keypair with:
#   python generate_key.py --signing-keypair --save
# This is the actual secret - keep it OFF the server and OFF any machine
# that ships to players. It only needs to exist wherever you run this
# script. The matching PUBLIC key goes into updater.py's SIGNING_PUBLIC_KEY
# instead of a shared secret, so decompiling updater.exe never exposes
# anything that could be used to forge a manifest (unlike a shared-secret
# HMAC key, which the client would also need to hold and could therefore
# leak from the exe).
# Read from a standalone .txt file (see _read_private_signing_key above),
# never from config.json - if that file is missing, signing is disabled
# (NOT recommended, see README).
SIGNING_PRIVATE_KEY = _read_private_signing_key(_CFG)

# Files the updater must never touch (launcher itself, local settings,
# self-update artifacts, etc.) - the self-update filenames are always
# ignored regardless of config.json, since scanning/hashing them as
# regular game files would never make sense. config.json's ignore.files
# is additive on top of this fixed set.
IGNORE_FILES = {
    MANIFEST_FILENAME,
    MANIFEST_FILENAME + ".sig",
    "updater.exe",                     # Your compiled updater executable
    "updater_manifest.json",           # Self-update manifest (see generate_updater_manifest.py)
    "updater_manifest.json.sig",
    "updater.exe.new",                 # Leftover from an in-progress self-update, if generate_manifest.py
    "updater.exe.update_helper.bat",   # is ever run inside a live client folder rather than a build dir
} | set(_ignore_cfg.get("files", ["Option.ini", "WindowsInfo.ini"]))

# Folders the updater must never scan or update
IGNORE_FOLDERS = set(_ignore_cfg.get("folders", ["screenshots", "logs", "replays"]))

# File extensions to always skip (lowercase, without the dot doesn't matter, include the dot)
IGNORE_EXTENSIONS = set(_ignore_cfg.get("extensions", [".log"]))

# ============================================================
# END CONFIGURATION
# ============================================================


def get_md5_bytes(data):
    """Computes the MD5 hash of an in-memory bytes object."""
    return hashlib.md5(data).hexdigest()


def get_md5(filepath):
    """Computes the MD5 hash of a file on disk."""
    hash_md5 = hashlib.md5()
    try:
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()
    except IOError:
        return None


def is_safe_rel_path(rel_path_web):
    """
    Defense in depth: refuses to put a path in the manifest that the
    updater's own sanitizer would reject anyway (absolute path, ".."
    segments, drive letters, backslashes). Should never trigger in
    practice since paths come from os.walk() on a real folder, but if
    it ever does, we want a loud failure here rather than a silently
    unusable manifest entry on the client.
    """
    if not rel_path_web or rel_path_web.startswith("/"):
        return False
    if ":" in rel_path_web or "\\" in rel_path_web:
        return False
    parts = rel_path_web.split("/")
    if any(p in ("", ".", "..") for p in parts):
        return False
    return True


def compress_file(filepath, compressed_path):
    """
    Compresses filepath into compressed_path with gzip. Every file is
    compressed unconditionally - if the result isn't actually smaller
    (common for already-compressed formats like .jpg/.mp3/.zip), the
    compressed copy is discarded and the manifest entry omits "compressed",
    so the server/updater fall back to serving the original file untouched.

    Uses mtime=0 and a GzipFile with no filename so the compressed output
    is byte-identical across runs when the input hasn't changed - otherwise
    gzip embeds the source mtime in its header and compressed_hash would
    change on every manifest regeneration even with no real content change.

    Returns (compressed_size, compressed_hash) or (None, None) if the
    compressed copy wasn't kept.
    """
    original_size = os.path.getsize(filepath)

    with open(filepath, "rb") as f_in:
        data = f_in.read()

    os.makedirs(os.path.dirname(compressed_path) or ".", exist_ok=True)

    with open(compressed_path, "wb") as raw_out:
        with gzip.GzipFile(fileobj=raw_out, mode="wb", compresslevel=COMPRESSION_LEVEL, mtime=0) as f_out:
            f_out.write(data)

    compressed_size = os.path.getsize(compressed_path)

    if compressed_size >= original_size:
        os.remove(compressed_path)
        return None, None

    compressed_hash = get_md5(compressed_path)
    return compressed_size, compressed_hash


def _compressed_path_for(client_dir, rel_path_web):
    """
    Where the .gz copy of rel_path_web should live on disk, honoring
    COMPRESSED_DIR (see its definition above for the "" vs "_gz" behavior).
    """
    gz_rel = rel_path_web + COMPRESSED_SUFFIX
    if COMPRESSED_DIR:
        return os.path.join(client_dir, COMPRESSED_DIR, *gz_rel.split("/"))
    return os.path.join(client_dir, *gz_rel.split("/"))


def _get_or_compress(filepath, rel_path_web, file_hash, client_dir, old_manifest, force_recompress):
    """
    Returns (compressed_size, compressed_hash, reused_bool).

    If force_recompress is False and the previous manifest already has an
    entry for this exact rel_path with the SAME source hash (i.e. the
    original file hasn't changed since the last run) AND a compressed
    copy that still actually exists on disk at the expected location -
    trusts the cached compressed_size/compressed_hash instead of
    re-running gzip. This is what makes repeated manifest generation on a
    mostly-unchanged client folder fast: only files that actually changed
    get recompressed.

    Any mismatch (hash differs, no cached entry, cached entry says
    "compressed": false, or the .gz file has gone missing from disk for
    any reason) falls through to actually compressing it fresh - caching
    is purely an optimization here, never a correctness requirement, so
    a torn/missing cache never produces a wrong manifest.
    """
    compressed_path = _compressed_path_for(client_dir, rel_path_web)

    if not force_recompress:
        old_entry = old_manifest.get(rel_path_web)
        if (
            isinstance(old_entry, dict)
            and old_entry.get("hash") == file_hash
            and old_entry.get("compressed") is True
            and old_entry.get("compressed_hash")
            and old_entry.get("compressed_size")
            and os.path.isfile(compressed_path)
        ):
            return old_entry["compressed_size"], old_entry["compressed_hash"], True

        if (
            isinstance(old_entry, dict)
            and old_entry.get("hash") == file_hash
            and old_entry.get("compressed") is False
        ):
            # Previously determined compression wasn't worth it for this
            # exact content (e.g. an already-compressed asset format) -
            # trust that too, rather than re-running gzip just to
            # rediscover the same "not worth it" result.
            return None, None, True

    compressed_size, compressed_hash = compress_file(filepath, compressed_path)
    return compressed_size, compressed_hash, False


def sign_manifest(manifest_bytes):
    """
    Returns a hex Ed25519 signature over the exact bytes written to
    manifest.json, or None if SIGNING_PRIVATE_KEY isn't configured. The
    updater verifies this signature with the matching PUBLIC key before
    trusting anything in the manifest, so a compromised/spoofed server
    can no longer hand out a manifest pointing files at arbitrary paths
    or vouching for a tampered game.exe - forging a valid signature
    requires this private key, which is never present in updater.exe or
    on the server, only wherever manifests are generated.
    """
    if not SIGNING_PRIVATE_KEY:
        return None
    private_key = Ed25519PrivateKey.from_private_bytes(bytes.fromhex(SIGNING_PRIVATE_KEY))
    signature = private_key.sign(manifest_bytes)
    return signature.hex()


def _iter_existing_gz_files(client_dir):
    """
    Yields absolute paths of every *.gz file currently on disk under
    client_dir, in both places a compressed copy could plausibly be:
    directly next to originals, and inside compressed_dir (if configured).
    Covers the "changed compressed_dir setting at some point" case too -
    stray .gz files from before the switch are still found and can be
    cleaned up rather than sitting there forever.
    """
    seen = set()

    # Direct-location .gz files (relevant when compressed_dir is "" now,
    # or was "" at some point in the past for files generated back then).
    for root, dirs, files in os.walk(client_dir):
        # Don't walk into compressed_dir here - it's handled by the
        # second pass below, and walking it here too would just mean
        # visiting the same files twice with different rel-path math.
        if COMPRESSED_DIR:
            compressed_dir_abs = os.path.realpath(os.path.join(client_dir, COMPRESSED_DIR))
            dirs[:] = [d for d in dirs if os.path.realpath(os.path.join(root, d)) != compressed_dir_abs]
        for file in files:
            if file.endswith(COMPRESSED_SUFFIX):
                path = os.path.join(root, file)
                if path not in seen:
                    seen.add(path)
                    yield path

    # compressed_dir's own .gz files (relevant when compressed_dir is set
    # now, or was set to a different value in the past - a rename of
    # compressed_dir leaves the old subfolder behind, which this also
    # cleans up since old_dir no longer matches COMPRESSED_DIR and so
    # isn't skipped by the os.walk above).
    if COMPRESSED_DIR:
        compressed_dir_path = os.path.join(client_dir, COMPRESSED_DIR)
        if os.path.isdir(compressed_dir_path):
            for root, _dirs, files in os.walk(compressed_dir_path):
                for file in files:
                    if file.endswith(COMPRESSED_SUFFIX):
                        path = os.path.join(root, file)
                        if path not in seen:
                            seen.add(path)
                            yield path


def clean_orphaned_gz_files(client_dir, manifest):
    """
    Removes .gz files on disk that no longer correspond to any entry in
    the freshly generated manifest - e.g. the original file was deleted
    or renamed, compression stopped being worth it for it, or
    compressed_dir was changed and old copies were left behind in the old
    location. Without this, compressed copies only ever accumulate: they
    get written when a file benefits from compression, but nothing ever
    deleted them again on its own.

    A .gz file is considered current (kept) only if BOTH:
      - its path matches exactly where _compressed_path_for() would put
        the compressed copy for some rel_path in the new manifest, AND
      - that manifest entry actually says "compressed": true (an entry
        that flipped to "not worth compressing" this run should have its
        old .gz removed even if the path would otherwise match).

    Returns the number of files removed. Never raises on a single file
    failing to delete (permissions, file in use, etc.) - logs and moves on,
    since a stale .gz left behind is a cosmetic problem, not a correctness
    one (the server/updater never trust a .gz that isn't in the manifest).
    """
    current_gz_paths = set()
    for rel_path_web, entry in manifest.items():
        if entry.get("compressed"):
            current_gz_paths.add(os.path.realpath(_compressed_path_for(client_dir, rel_path_web)))

    removed = 0
    for gz_path in _iter_existing_gz_files(client_dir):
        if os.path.realpath(gz_path) in current_gz_paths:
            continue
        try:
            os.remove(gz_path)
            removed += 1
        except OSError as e:
            print(f"  WARNING: couldn't remove orphaned {gz_path}: {e}")

    # Clean up now-empty directories left behind inside compressed_dir
    # (e.g. a whole subfolder of the game was removed). Doesn't touch
    # compressed_dir's own top-level folder even if it ends up empty -
    # leaving an empty "_gz" around is harmless and avoids surprising
    # "the folder I configured just vanished" behavior.
    if COMPRESSED_DIR:
        compressed_dir_path = os.path.join(client_dir, COMPRESSED_DIR)
        if os.path.isdir(compressed_dir_path):
            for root, dirs, files in os.walk(compressed_dir_path, topdown=False):
                if root == compressed_dir_path:
                    continue
                if not dirs and not files:
                    try:
                        os.rmdir(root)
                    except OSError:
                        pass

    return removed


def generate(client_dir, force_recompress=False, clean_orphans=True):
    if not os.path.exists(client_dir) or not os.path.isdir(client_dir):
        print(f"Error: directory '{client_dir}' not found!")
        return

    manifest = {}
    manifest_path = os.path.join(client_dir, MANIFEST_FILENAME)
    sig_path = manifest_path + ".sig"

    # Load the previous manifest (if any) so unchanged files can skip
    # recompression - see _get_or_compress below. A missing/corrupt old
    # manifest just means nothing is cached yet, not an error.
    old_manifest = {}
    if os.path.exists(manifest_path):
        try:
            with open(manifest_path, "r", encoding="utf-8") as f:
                old_manifest = json.load(f)
        except (OSError, json.JSONDecodeError):
            old_manifest = {}

    compressed_dir_abs = os.path.realpath(os.path.join(client_dir, COMPRESSED_DIR)) if COMPRESSED_DIR else None

    print(f"Scanning client folder: {client_dir}")
    if force_recompress:
        print("--force-recompress given: recompressing every file regardless of cache.")

    total_original = 0
    total_saved = 0
    skipped_unsafe = 0
    reused_count = 0

    for root, dirs, files in os.walk(client_dir):
        # Filter out ignored folders (case-insensitive)...
        dirs[:] = [d for d in dirs if d.lower() not in IGNORE_FOLDERS]

        # ...and specifically the compressed_dir subtree, wherever it is
        # nested, so its .gz files are never themselves scanned as game
        # files. IGNORE_FOLDERS above only matches by folder *name* at any
        # depth; this matches the exact configured path, since a name
        # like "_gz" could in principle collide with an unrelated folder
        # the game legitimately ships.
        if compressed_dir_abs is not None:
            dirs[:] = [
                d for d in dirs
                if os.path.realpath(os.path.join(root, d)) != compressed_dir_abs
            ]

        for file in files:
            if file in IGNORE_FILES:
                continue
            if any(file.lower().endswith(ext) for ext in IGNORE_EXTENSIONS):
                continue
            # Skip stray leftover compressed copies from a previous run
            # (relevant when compressed_dir is "" and .gz files sit right
            # next to their originals) so we never treat "texture.dds.gz"
            # as a distributable file itself.
            if file.endswith(COMPRESSED_SUFFIX):
                continue

            filepath = os.path.join(root, file)

            # Path relative to the client folder (this is what the updater expects)
            rel_path = os.path.relpath(filepath, client_dir)
            rel_path_web = rel_path.replace("\\", "/")

            if not is_safe_rel_path(rel_path_web):
                print(f"  WARNING: skipping unsafe path (should never happen): {rel_path_web}")
                skipped_unsafe += 1
                continue

            file_hash = get_md5(filepath)
            if not file_hash:
                continue

            entry = {
                "hash": file_hash,
                "size": os.path.getsize(filepath),
            }

            compressed_size, compressed_hash, reused = _get_or_compress(
                filepath, rel_path_web, file_hash, client_dir,
                old_manifest, force_recompress,
            )
            if reused:
                reused_count += 1

            if compressed_hash:
                entry["compressed"] = True
                entry["compressed_size"] = compressed_size
                entry["compressed_hash"] = compressed_hash
                total_saved += entry["size"] - compressed_size
            else:
                entry["compressed"] = False

            total_original += entry["size"]
            manifest[rel_path_web] = entry

    try:
        # Serialize once with sorted, fixed formatting so the byte sequence
        # that gets hashed/signed is deterministic and matches exactly what
        # gets written to disk (and what the updater will read back).
        manifest_bytes = json.dumps(manifest, indent=4, sort_keys=True).encode("utf-8")

        with open(manifest_path, "wb") as f:
            f.write(manifest_bytes)

        signature = sign_manifest(manifest_bytes)
        if signature:
            with open(sig_path, "w", encoding="utf-8") as f:
                f.write(signature)
            print(f"Manifest signed (Ed25519). Signature saved to: {sig_path}")
        else:
            print("WARNING: no private signing key found - manifest.json is UNSIGNED.")
            print("         Run: python generate_key.py --signing-keypair --save")
            print("         The updater will refuse unsigned manifests unless it")
            print("         has REQUIRE_SIGNATURE = False, which is not recommended.")
            if os.path.exists(sig_path):
                os.remove(sig_path)  # don't leave a stale signature lying around

        print(f"Done! Manifest saved to: {manifest_path}")
        print(f"Total files in manifest: {len(manifest)}")
        if skipped_unsafe:
            print(f"Skipped {skipped_unsafe} file(s) with unsafe paths.")
        if total_original:
            pct = (total_saved / total_original) * 100
            print(f"Compression saved {total_saved:,} bytes of {total_original:,} ({pct:.1f}%)")

        if clean_orphans:
            removed = clean_orphaned_gz_files(client_dir, manifest)
            if removed:
                print(f"Removed {removed} orphaned .gz file(s) no longer referenced by the manifest.")
    except IOError as e:
        print(f"Error saving manifest: {e}")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Universal manifest generator for game updaters.")
    parser.add_argument("client_path", help="Path to the client files folder")
    parser.add_argument(
        "--force-recompress", action="store_true",
        help=(
            "Recompress every file with gzip, ignoring the cache from the previous "
            "manifest.json (which normally skips recompression for files whose hash "
            "hasn't changed). Use this after changing manifest.compression_level in "
            "config.json, or if you suspect the .gz cache is stale/corrupt."
        ),
    )
    parser.add_argument(
        "--no-clean-orphans", action="store_true",
        help=(
            "Don't remove .gz files that no longer correspond to any file in the "
            "manifest (leftover from deleted/renamed originals, or from switching "
            "manifest.compressed_dir in config.json). Cleanup runs by default."
        ),
    )

    args = parser.parse_args()

    generate(
        args.client_path,
        force_recompress=args.force_recompress,
        clean_orphans=not args.no_clean_orphans,
    )