poison-flower/simple-updater
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fixes

Browse Source
This commit is contained in:
poison-flower
2026-09-12 03:11:49 +03:00
parent b184ea76f4
commit a71580462f
11 changed files with 2303 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
updater.py
+521 -11
View File
@@ -3,6 +3,7 @@ import sys
import json
import gzip
import hashlib
import tempfile
import threading
import urllib.request
import urllib.parse
@@ -11,6 +12,9 @@ import configparser
import tkinter as tk
from tkinter import ttk, messagebox
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.exceptions import InvalidSignature
try:
from PIL import Image, ImageTk
HAS_PIL = True
@@ -19,6 +23,24 @@ except ImportError:
# ============================================================
# CONFIGURATION - edit these values for your game/project
#
# These are the values baked into a compiled updater.exe. build_client.py
# writes this whole block (pulling values from a running server's
# /client_build_info.json, not from a local config.json - see
# build_client.py's docstring) before running PyInstaller, so in normal
# use you edit the SERVER's config.json and re-run build_client.py rather
# than touching this file by hand. See config.json / README for what each
# setting under updater_ui / connection / paths / manifest / self_update /
# signing_keys / access_keys maps to here.
#
# The block below is also read as a live fallback: if this script is run
# directly with `python updater.py` (not the compiled exe) and a
# config.json happens to sit next to it, values from config.json override
# these constants at import time - convenient for local testing without
# recompiling. A compiled exe never does this lookup (see
# CONFIG_OVERRIDES_APPLIED below) - the values below are simply what got
# baked in at build time, so a released exe never depends on a
# config.json existing on the player's machine.
# ============================================================
GAME_NAME = "My Game" # Shown in the window title
@@ -54,6 +76,33 @@ CLIENT_KEYS = {
}
DEFAULT_CLIENT = "Main"
# Ed25519 PUBLIC key used to verify manifest.json's signature
# (manifest.json.sig). Must match the public half of the keypair whose
# PRIVATE half is used by generate_manifest.py's SIGNING_PRIVATE_KEY.
# Generate a keypair with:
# python generate_key.py --signing-keypair
#
# This is what actually protects players: even if the server is
# compromised, spoofed via DNS, or MITM'd over plain HTTP, the attacker
# cannot produce a manifest the updater will accept without the matching
# PRIVATE key - which never ships in this exe. Unlike a shared-secret
# (HMAC) scheme, this public key is not sensitive: decompiling updater.exe
# and recovering this value gives an attacker nothing they can use to
# forge a signature, only to verify one (which they could already do by
# just running the updater).
SIGNING_PUBLIC_KEY = None # e.g. "b3a1...64 hex chars (32 bytes)..."
# If True, the updater refuses to install anything from an unsigned or
# badly-signed manifest. Turning this off makes the whole signature
# mechanism pointless - only do this for local testing.
REQUIRE_SIGNATURE = True
# Hard cap on how large a single decompressed file is allowed to be before
# the updater gives up on the compressed copy and falls back to the plain
# download. Prevents a hostile/compromised server from serving a tiny .gz
# that decompresses into gigabytes and exhausts memory ("zip bomb").
MAX_DECOMPRESSED_SIZE = 512 * 1024 * 1024 # 512 MiB, adjust for your game
# Relative path (from the game folder) to the executable that launches the game
GAME_EXE_RELATIVE_PATH = os.path.join("system", "game.exe")
@@ -62,6 +111,7 @@ BANNER_RELATIVE_PATH = "banner.png"
ICON_RELATIVE_PATH = "icon.ico"
MANIFEST_FILENAME = "manifest.json"
SIGNATURE_FILENAME = MANIFEST_FILENAME + ".sig"
# Suffix used for precompressed files on the server (must match
# COMPRESSED_SUFFIX in generate_manifest.py).
@@ -70,10 +120,116 @@ COMPRESSED_SUFFIX = ".gz"
# Local settings file (created next to the updater on first run).
SETTINGS_FILENAME = "updater.ini"
# ------------------------------------------------------------
# Self-update (the updater updates itself, not just the game)
# ------------------------------------------------------------
# Bump this every time you rebuild updater.exe. Must match the "version"
# field you put in updater_manifest.json when you publish a new build.
# Purely a monotonically increasing integer - the updater does not try to
# parse semver or anything fancier than "is the server's number bigger
# than mine".
UPDATER_VERSION = 1
# Name of the self-update manifest on the server, fetched from the SAME
# access-key folder as the game manifest (so it inherits the same routing
# and the same signature verification - see _fetch_verified_json below).
# Deliberately a separate file from manifest.json: which updater build is
# current has nothing to do with which game-client version the player
# picked (Main/Test), so it shouldn't be duplicated per client folder or
# accidentally tied to game file hashes.
UPDATER_MANIFEST_FILENAME = "updater_manifest.json"
UPDATER_MANIFEST_SIGNATURE_FILENAME = UPDATER_MANIFEST_FILENAME + ".sig"
# If True, checks for a new updater.exe once per run, before checking game
# files. If a newer signed build is available, the updater replaces itself
# and relaunches - the game-file update is deferred to the *next* run of
# the new exe, so a self-update never happens in the middle of an
# in-progress game-file update.
ENABLE_SELF_UPDATE = True
# ============================================================
# END CONFIGURATION
# ============================================================
# ------------------------------------------------------------
# Optional live override from config.json - dev/test convenience only.
#
# This block NEVER runs inside a compiled exe (sys.frozen is True there,
# so it's skipped outright) - a released exe's behavior is fully decided
# by the constants above, baked in at build time by quickstart.py. This
# exists purely so `python updater.py` picks up config.json changes
# during local testing without needing a rebuild each time.
#
# Deliberately reads only signing_keys.public - config.json never
# contains a private key at all (it lives in a standalone .txt file that
# only generate_manifest.py / generate_updater_manifest.py touch), so
# there's no "private" field here to accidentally read even by mistake.
# This block only pulls the subset of config.json that's safe/meaningful
# for a client to know, mirroring exactly what quickstart.py bakes into a
# real build.
# ------------------------------------------------------------
CONFIG_OVERRIDES_APPLIED = False
if not getattr(sys, "frozen", False):
_config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.json")
if os.path.exists(_config_path):
try:
with open(_config_path, "r", encoding="utf-8") as _f:
_cfg = json.load(_f)
_ui = _cfg.get("updater_ui", {})
GAME_NAME = _ui.get("game_name", GAME_NAME)
UPDATER_NAME = _ui.get("updater_name", UPDATER_NAME)
WINDOW_WIDTH = _ui.get("window_width", WINDOW_WIDTH)
WINDOW_HEIGHT = _ui.get("window_height", WINDOW_HEIGHT)
BANNER_WIDTH = _ui.get("banner_width", BANNER_WIDTH)
BANNER_HEIGHT = _ui.get("banner_height", BANNER_HEIGHT)
_conn = _cfg.get("connection", {})
SERVER_ADDRESS = _conn.get("server_address", SERVER_ADDRESS)
SERVER_PORT = str(_conn.get("server_port", SERVER_PORT))
USE_HTTPS = _conn.get("use_https", USE_HTTPS)
_access_keys = _cfg.get("access_keys", {})
if _access_keys:
CLIENT_KEYS = {name: key for name, key in _access_keys.items() if key}
DEFAULT_CLIENT = _cfg.get("default_client", DEFAULT_CLIENT)
_signing = _cfg.get("signing_keys", {})
SIGNING_PUBLIC_KEY = _signing.get("public") or SIGNING_PUBLIC_KEY
REQUIRE_SIGNATURE = _cfg.get("require_signature", REQUIRE_SIGNATURE)
_paths = _cfg.get("paths", {})
GAME_EXE_RELATIVE_PATH = os.path.normpath(
_paths.get("game_exe_relative_path", GAME_EXE_RELATIVE_PATH)
)
BANNER_RELATIVE_PATH = _paths.get("banner_relative_path", BANNER_RELATIVE_PATH)
ICON_RELATIVE_PATH = _paths.get("icon_relative_path", ICON_RELATIVE_PATH)
SETTINGS_FILENAME = _paths.get("settings_filename", SETTINGS_FILENAME)
_manifest = _cfg.get("manifest", {})
MANIFEST_FILENAME = _manifest.get("manifest_filename", MANIFEST_FILENAME)
SIGNATURE_FILENAME = MANIFEST_FILENAME + ".sig"
COMPRESSED_SUFFIX = _manifest.get("compressed_suffix", COMPRESSED_SUFFIX)
MAX_DECOMPRESSED_SIZE = int(
_manifest.get("max_decompressed_size_mb", MAX_DECOMPRESSED_SIZE // (1024 * 1024))
) * 1024 * 1024
_self_update = _cfg.get("self_update", {})
ENABLE_SELF_UPDATE = _self_update.get("enabled", ENABLE_SELF_UPDATE)
UPDATER_VERSION = _self_update.get("updater_version", UPDATER_VERSION)
UPDATER_MANIFEST_FILENAME = _self_update.get(
"updater_manifest_filename", UPDATER_MANIFEST_FILENAME
)
UPDATER_MANIFEST_SIGNATURE_FILENAME = UPDATER_MANIFEST_FILENAME + ".sig"
CONFIG_OVERRIDES_APPLIED = True
except (OSError, json.JSONDecodeError, AttributeError, TypeError):
# Malformed config.json during dev testing shouldn't crash the
# app - just fall back to the constants above.
pass
def get_resource_path(relative_path):
if getattr(sys, 'frozen', False):
@@ -94,6 +250,12 @@ GAME_EXE = os.path.join(BASE_PATH, GAME_EXE_RELATIVE_PATH)
SETTINGS_FILE = os.path.join(BASE_PATH, SETTINGS_FILENAME)
class ManifestSecurityError(Exception):
"""Raised when the manifest fails signature verification or contains
an unsafe path. Always aborts the update - never partially applied."""
pass
def load_settings():
"""
Reads settings.ini next to the updater. If it doesn't exist yet (first
@@ -134,6 +296,82 @@ def build_server_base_url():
return f"{scheme}://{SERVER_ADDRESS}:{SERVER_PORT}/"
def verify_manifest_signature(manifest_bytes, signature_hex):
"""
Verifies manifest_bytes against signature_hex using Ed25519 and
SIGNING_PUBLIC_KEY. Returns True/False.
If SIGNING_PUBLIC_KEY isn't configured, returns True only when
REQUIRE_SIGNATURE is False (test/dev mode) - this keeps "no key
configured" from silently behaving as "always trusted" in a real
deployment.
Unlike an HMAC scheme, this key is asymmetric: SIGNING_PUBLIC_KEY
can only verify signatures, never produce them. So even a full
decompilation of this exe cannot yield anything usable to forge a
manifest - the private key needed for that never ships here.
"""
if not SIGNING_PUBLIC_KEY:
return not REQUIRE_SIGNATURE
if not signature_hex:
return False
try:
public_key = Ed25519PublicKey.from_public_bytes(bytes.fromhex(SIGNING_PUBLIC_KEY))
public_key.verify(bytes.fromhex(signature_hex.strip()), manifest_bytes)
return True
except (InvalidSignature, ValueError):
return False
def sanitize_manifest_rel_path(rel_path):
"""
Validates a path taken from the (now-verified) manifest before it's
ever joined with BASE_PATH and written to disk. This is the fix for
the original path-traversal hole: previously os.path.normpath() alone
did NOT stop rel_path values like "../../whatever" or an absolute
path ("C:\\Windows\\..." or "/etc/...") from escaping the game folder
entirely, because normpath() only collapses redundant separators/dots -
it does not confine the result to any base directory.
Returns the normalized, safe, OS-specific relative path, or raises
ManifestSecurityError if the path tries to escape BASE_PATH in any way.
"""
if not rel_path or not isinstance(rel_path, str):
raise ManifestSecurityError(f"Invalid path entry: {rel_path!r}")
# Manifest paths are always forward-slash (see generate_manifest.py).
# Reject backslashes and drive letters outright - on Windows a value
# like "C:\\Windows\\System32\\x.dll" or "\\\\server\\share" must never
# reach os.path.join() as a "relative" path.
if "\\" in rel_path or ":" in rel_path:
raise ManifestSecurityError(f"Unsafe path in manifest: {rel_path!r}")
if rel_path.startswith("/"):
raise ManifestSecurityError(f"Absolute path in manifest: {rel_path!r}")
parts = rel_path.split("/")
if any(p in ("", ".", "..") for p in parts):
raise ManifestSecurityError(f"Path traversal attempt in manifest: {rel_path!r}")
local_path = os.path.normpath(os.path.join(BASE_PATH, *parts))
# Belt-and-braces: after normalization, confirm the result is still
# actually inside BASE_PATH. Catches anything the segment-level check
# above might have missed (e.g. platform-specific quirks).
base_real = os.path.realpath(BASE_PATH)
target_real = os.path.realpath(local_path)
try:
if os.path.commonpath([target_real, base_real]) != base_real:
raise ManifestSecurityError(f"Path escapes game folder: {rel_path!r}")
except ValueError:
# Different drives on Windows - definitely not inside BASE_PATH.
raise ManifestSecurityError(f"Path escapes game folder: {rel_path!r}")
return local_path
class UpdaterApp:
def __init__(self, root):
self.root = root
@@ -259,6 +497,213 @@ class UpdaterApp:
self.mode_combo.config(state="disabled") # Lock the selector during update
threading.Thread(target=self.update_process, daemon=True).start()
def _fetch_verified_json(self, base_url, filename, signature_filename):
"""
Downloads <filename> + <signature_filename> from base_url, verifies
the Ed25519 signature over the raw manifest bytes, and only then
parses/returns it as JSON. Shared by both the game-file manifest
and the self-update manifest - same trust model, same failure mode
(raise ManifestSecurityError, never partially trust the content).
Raises ManifestSecurityError if verification fails (or is required
but the signature is missing). Raises normal exceptions (URLError,
json errors, ...) for ordinary network/parsing failures - callers
distinguish the two so a plain network hiccup doesn't get reported
as a security incident.
"""
data_url = base_url + filename
sig_url = base_url + signature_filename
req = urllib.request.urlopen(data_url, timeout=10)
raw_bytes = req.read()
signature_hex = None
try:
sig_req = urllib.request.urlopen(sig_url, timeout=10)
signature_hex = sig_req.read().decode("utf-8", errors="ignore").strip()
except urllib.error.HTTPError:
signature_hex = None # No .sig on server - handled below.
if not verify_manifest_signature(raw_bytes, signature_hex):
if SIGNING_PUBLIC_KEY and REQUIRE_SIGNATURE:
raise ManifestSecurityError(
f"{filename}: signature verification failed - refusing to use it. "
"The server may be compromised, or the signing keys are out of sync."
)
if REQUIRE_SIGNATURE:
raise ManifestSecurityError(
f"No SIGNING_PUBLIC_KEY configured but REQUIRE_SIGNATURE is True - refusing {filename}."
)
# REQUIRE_SIGNATURE is False and no key configured: dev/test mode.
return json.loads(raw_bytes.decode("utf-8"))
def _fetch_verified_manifest(self, current_server_url):
return self._fetch_verified_json(current_server_url, MANIFEST_FILENAME, SIGNATURE_FILENAME)
def _check_self_update(self, current_server_url):
"""
Checks updater_manifest.json (signed the same way as the game
manifest) for a newer build of this very updater.exe. If one is
available, downloads it, verifies its hash, and hands off to a
tiny helper script that does the actual file swap - because a
running Windows executable cannot overwrite or delete its own
.exe file while it's still executing.
Returns True if a self-update was launched (caller should stop
immediately and let the new process take over), False if there's
nothing to do (already current, feature disabled, or the check/
download failed - any failure here is non-fatal: it just means
we continue with the regular game-file update on this old build).
"""
if not ENABLE_SELF_UPDATE:
return False
if not getattr(sys, "frozen", False):
# Running as a plain .py script (dev mode) - there's no exe to
# replace, and re-launching "python updater.py" from a batch
# script would be more surprising than helpful. Self-update
# only makes sense for the compiled artifact.
return False
try:
self.update_ui(text="Checking for launcher updates...")
update_info = self._fetch_verified_json(
current_server_url, UPDATER_MANIFEST_FILENAME, UPDATER_MANIFEST_SIGNATURE_FILENAME
)
except ManifestSecurityError:
# Re-raise: an updater_manifest.json that exists but fails
# signature verification is treated as seriously as a game
# manifest that fails verification - it's the same signal
# (compromised/spoofed server) and should stop the update the
# same way, not be silently swallowed.
raise
except Exception:
# Ordinary network/parsing failure (server doesn't have the
# file yet, timeout, etc.) - not a security event, just skip
# self-update this run and continue with the game update.
return False
remote_version = update_info.get("version")
download_name = update_info.get("filename", "updater.exe")
expected_hash = update_info.get("hash")
if not isinstance(remote_version, int) or remote_version <= UPDATER_VERSION:
return False # Already current (or manifest is malformed - fail closed).
if not expected_hash:
return False # Malformed entry - fail closed rather than install unverified.
# download_name is attacker-influenced content (it comes from the
# server, same trust boundary as the game manifest's rel_path) -
# route it through the exact same sanitizer used for game files
# instead of trusting it to be a bare filename.
try:
sanitize_manifest_rel_path(download_name)
except ManifestSecurityError:
return False
try:
self.update_ui(text=f"Downloading launcher update v{remote_version}...")
current_exe = sys.executable
new_exe_path = current_exe + ".new"
exe_url = current_server_url + urllib.parse.quote(download_name, safe='/')
with urllib.request.urlopen(exe_url, timeout=120) as resp:
new_exe_data = resp.read(MAX_DECOMPRESSED_SIZE + 1)
if len(new_exe_data) > MAX_DECOMPRESSED_SIZE:
return False
if self.get_md5_bytes(new_exe_data) != expected_hash:
return False # Corrupt or tampered download - don't install it.
self._atomic_write(new_exe_path, new_exe_data)
self._launch_self_update_helper(current_exe, new_exe_path)
return True
except Exception:
# Any failure in the download/swap path - clean up and fall
# back to running the current build. Never leaves the update
# half-applied: the swap itself only happens in the helper
# script below, after this process has already exited.
try:
stray_new_path = sys.executable + ".new"
if os.path.exists(stray_new_path):
os.remove(stray_new_path)
except OSError:
pass
return False
def _launch_self_update_helper(self, current_exe, new_exe_path):
"""
Writes a small batch script next to the exe and launches it
detached, then this process should exit immediately (the caller
does that). The script:
1. Waits for this process to actually exit (a running Windows
exe can't be replaced in place - "del" on it fails with a
sharing violation until the handle is released), retrying
for a bounded number of attempts rather than looping forever.
2. Swaps updater.exe.new into place over updater.exe.
3. Relaunches the new exe.
4. Deletes itself (a batch file CAN delete itself as its last
action on Windows, since the OS only locks the file while a
command *within* it is being read, not the whole script).
This is Windows-only (matches the rest of the codebase's
os.startfile()/PyInstaller --onefile assumptions). On non-Windows
this is a no-op; self-update there would need a different
mechanism (e.g. exec() with a swapped path) that isn't
implemented here.
"""
if os.name != "nt":
return # Unsupported platform for this helper - caller treats as failure.
pid = os.getpid()
helper_path = current_exe + ".update_helper.bat"
# Quote paths for cmd.exe; exe paths under Program Files etc. can
# contain spaces.
script = (
"@echo off\r\n"
"setlocal\r\n"
f'set "CUR={current_exe}"\r\n'
f'set "NEW={new_exe_path}"\r\n'
f'set "PID={pid}"\r\n'
"set /a TRIES=0\r\n"
":waitloop\r\n"
"set /a TRIES+=1\r\n"
"if %TRIES% GTR 100 goto giveup\r\n"
# tasklist exits 0 and prints a row if the PID is still alive;
# findstr returns non-zero (errorlevel 1) if it finds nothing,
# which is the "process has exited, safe to proceed" case.
f'tasklist /FI "PID eq %PID%" 2^>nul | findstr /C:"%PID%" >nul\r\n'
"if %ERRORLEVEL%==0 (\r\n"
" timeout /t 1 /nobreak >nul\r\n"
" goto waitloop\r\n"
")\r\n"
'move /y "%NEW%" "%CUR%" >nul\r\n'
'start "" "%CUR%"\r\n'
"goto cleanup\r\n"
":giveup\r\n"
# Couldn't confirm the old process exited after ~100 tries -
# leave .new in place rather than risk swapping over a file
# that's still in use, and don't relaunch anything.
":cleanup\r\n"
'(goto) 2>nul & del "%~f0"\r\n'
)
with open(helper_path, "w", encoding="ascii", newline="") as f:
f.write(script)
# DETACHED_PROCESS + CREATE_NEW_PROCESS_GROUP so the helper survives
# this process exiting and isn't tied to our console/job.
DETACHED_PROCESS = 0x00000008
CREATE_NEW_PROCESS_GROUP = 0x00000200
subprocess.Popen(
["cmd.exe", "/c", helper_path],
creationflags=DETACHED_PROCESS | CREATE_NEW_PROCESS_GROUP,
close_fds=True,
)
def update_process(self):
# Build URLs dynamically based on the selected client version.
# The display name is only for the UI - the actual URL uses the
@@ -266,12 +711,20 @@ class UpdaterApp:
display_name = self.mode_var.get()
access_key = CLIENT_KEYS[display_name]
current_server_url = build_server_base_url() + f"{access_key}/"
manifest_url = current_server_url + MANIFEST_FILENAME
try:
self.update_ui(text=f"Connecting to server ({display_name})...")
req = urllib.request.urlopen(manifest_url, timeout=10)
server_manifest = json.loads(req.read().decode('utf-8'))
if self._check_self_update(current_server_url):
# A new updater.exe is downloaded and the swap-and-relaunch
# helper is already launched (detached). This process must
# exit now without touching game files - the freshly
# relaunched exe will check game files on its own next.
self.update_ui(text="Launcher update found - restarting...")
self.root.after(300, self.root.destroy)
return
server_manifest = self._fetch_verified_manifest(current_server_url)
files_to_download = []
total_files = len(server_manifest)
@@ -280,7 +733,11 @@ class UpdaterApp:
for i, (rel_path, entry) in enumerate(server_manifest.items()):
self.update_ui(text=f"Checking: {rel_path}", prog_val=i)
local_path = os.path.join(BASE_PATH, os.path.normpath(rel_path))
# Validate BEFORE ever joining this into a filesystem path.
# A malicious/compromised server could otherwise use a key
# like "../../../Windows/System32/x.dll" to write outside
# the game folder entirely.
local_path = sanitize_manifest_rel_path(rel_path)
local_hash = self.get_md5(local_path)
if local_hash != entry["hash"]:
@@ -294,7 +751,7 @@ class UpdaterApp:
self.update_ui(text=f"Downloading ({i + 1}/{total_dl}): {rel_path}", prog_val=i)
entry = server_manifest[rel_path]
local_path = os.path.join(BASE_PATH, os.path.normpath(rel_path))
local_path = sanitize_manifest_rel_path(rel_path)
os.makedirs(os.path.dirname(local_path), exist_ok=True)
self._download_file(current_server_url, rel_path, entry, local_path)
@@ -304,18 +761,52 @@ class UpdaterApp:
self.root.after(0, lambda: self.btn_update.config(state="normal"))
self.root.after(0, lambda: self.mode_combo.config(state="readonly"))
except ManifestSecurityError as e:
# Security failures get their own message so they're visibly
# distinct from ordinary network hiccups - this should never
# be silently retried or ignored.
self.update_ui(text=f"Security error: {str(e)}")
self.root.after(0, lambda: messagebox.showerror("Update blocked", str(e)))
self.root.after(0, lambda: self.btn_update.config(state="normal"))
self.root.after(0, lambda: self.mode_combo.config(state="readonly"))
except Exception as e:
self.update_ui(text=f"Error: {str(e)}")
self.root.after(0, lambda: self.btn_update.config(state="normal"))
self.root.after(0, lambda: self.mode_combo.config(state="readonly"))
def _atomic_write(self, local_path, data):
"""
Writes data to local_path via a temp file + os.replace() in the
same directory, so a crash or power loss mid-write can never leave
a half-written file sitting at the real path (which previously
could be picked up as "the file" on next launch, or briefly exist
in a corrupt state if the game happened to be running concurrently).
os.replace() is atomic on both Windows and POSIX when source and
destination are on the same filesystem, which they always are here
since the temp file is created right next to the target.
"""
directory = os.path.dirname(local_path) or "."
fd, tmp_path = tempfile.mkstemp(dir=directory, prefix=".tmp_upd_")
try:
with os.fdopen(fd, "wb") as f:
f.write(data)
os.replace(tmp_path, local_path)
except Exception:
try:
os.remove(tmp_path)
except OSError:
pass
raise
def _download_file(self, base_url, rel_path, entry, local_path):
"""
Downloads one file. If the manifest marks it as compressed, fetches
the .gz copy, verifies its hash, and decompresses it into local_path.
Otherwise downloads the original file directly. Falls back to the
uncompressed original if the compressed copy fails verification or
download, so a bad/missing .gz on the server doesn't block the update.
uncompressed original if the compressed copy fails verification,
download, or is too large once decompressed - so a bad/missing/
oversized .gz on the server doesn't block the update and can't be
used as a memory-exhaustion ("zip bomb") vector.
"""
if entry.get("compressed"):
gz_url = base_url + urllib.parse.quote(rel_path + COMPRESSED_SUFFIX, safe='/')
@@ -326,13 +817,24 @@ class UpdaterApp:
if self.get_md5_bytes(compressed_data) != entry.get("compressed_hash"):
raise ValueError("compressed file hash mismatch")
data = gzip.decompress(compressed_data)
# Guard against decompression bombs: decompress with a hard
# byte cap instead of gzip.decompress()'s unbounded
# "read everything into memory" behavior. If the real file
# is legitimately larger than the cap for your game, raise
# MAX_DECOMPRESSED_SIZE in the config block above.
expected_size = entry.get("size", 0)
if expected_size > MAX_DECOMPRESSED_SIZE:
raise ValueError("file exceeds MAX_DECOMPRESSED_SIZE, refusing to decompress")
decompressor = gzip.GzipFile(fileobj=__import__("io").BytesIO(compressed_data))
data = decompressor.read(MAX_DECOMPRESSED_SIZE + 1)
if len(data) > MAX_DECOMPRESSED_SIZE:
raise ValueError("decompressed data exceeded MAX_DECOMPRESSED_SIZE")
if self.get_md5_bytes(data) != entry["hash"]:
raise ValueError("decompressed file hash mismatch")
with open(local_path, "wb") as f:
f.write(data)
self._atomic_write(local_path, data)
return
except Exception:
# Fall through to downloading the uncompressed original below.
@@ -340,7 +842,15 @@ class UpdaterApp:
safe_url_path = urllib.parse.quote(rel_path, safe='/')
file_url = base_url + safe_url_path
urllib.request.urlretrieve(file_url, local_path)
with urllib.request.urlopen(file_url, timeout=60) as resp:
data = resp.read(MAX_DECOMPRESSED_SIZE + 1)
if len(data) > MAX_DECOMPRESSED_SIZE:
raise ValueError(f"{rel_path} exceeds MAX_DECOMPRESSED_SIZE, aborting download")
if self.get_md5_bytes(data) != entry["hash"]:
raise ValueError(f"{rel_path}: downloaded file hash mismatch")
self._atomic_write(local_path, data)
def play_game(self):
if os.path.exists(GAME_EXE):