#!/usr/bin/env python3 """Validate and stage one replay plus its exact map into a private session.""" from __future__ import annotations import argparse import hashlib import json import os import shutil import tempfile from pathlib import Path MAX_REPLAY_BYTES = 64 * 1024 * 1024 MAX_MAP_BYTES = 256 * 1024 * 1024 def sha1_file(path: Path, limit: int) -> tuple[str, int]: digest = hashlib.sha1() size = 0 with path.open("rb") as source: while chunk := source.read(1024 * 1024): size += len(chunk) if size > limit: raise ValueError(f"file exceeds {limit} bytes") digest.update(chunk) return digest.hexdigest(), size def safe_child(root: Path, value: str) -> Path: if not isinstance(value, str) or not value or "\0" in value: raise ValueError("invalid relative file") relative = Path(value) if relative.is_absolute() or any(part in ("", ".", "..") for part in relative.parts): raise ValueError("relative file escapes its root") result = (root / relative).resolve() if os.path.commonpath((str(root), str(result))) != str(root): raise ValueError("relative file escapes its root") return result def map_parts(value: object, allow_expired_patch_map: bool = False) -> list[str]: if not isinstance(value, str) or len(value) > 512 or ":" in value: raise ValueError("invalid map path") parts = value.replace("/", "\\").split("\\") if (len(parts) < 2 or parts[0].lower() != "maps" or any(part in ("", ".", "..") for part in parts)): raise ValueError("map path must be below Maps") parts[0] = "Maps" extensions = (".w3m", ".w3x", ".tmp") if allow_expired_patch_map \ else (".w3m", ".w3x") if Path(parts[-1]).suffix.lower() not in extensions: raise ValueError("map path has an unsupported extension") return parts def install(source: Path, destination: Path, digest: str, limit: int) -> None: destination.parent.mkdir(parents=True, exist_ok=True) if destination.is_file() and not destination.is_symlink(): actual, size = sha1_file(destination, limit) if size and actual == digest: return descriptor, temporary_name = tempfile.mkstemp( prefix=".w3cs-", suffix=".tmp", dir=destination.parent) temporary = Path(temporary_name) try: with source.open("rb") as input_file, os.fdopen( descriptor, "wb") as output_file: shutil.copyfileobj(input_file, output_file, 1024 * 1024) output_file.flush() os.fsync(output_file.fileno()) actual, size = sha1_file(temporary, limit) if not size or actual != digest: raise ValueError("staged file identity differs") os.chmod(temporary, 0o644) os.replace(temporary, destination) finally: temporary.unlink(missing_ok=True) def stage(metadata_path: Path, replay: Path, prefix: Path, game_directory: Path, replay_directory: str = "Replays") -> dict[str, object]: metadata_path = metadata_path.resolve() replay_root = metadata_path.parent replay = replay.resolve() prefix = prefix.resolve() game_directory = game_directory.resolve() if os.path.commonpath((str(replay_root), str(replay))) != str(replay_root): raise ValueError("replay escapes its catalog root") metadata = json.loads(metadata_path.read_text(encoding="utf-8")) if not isinstance(metadata, dict): raise ValueError("replay metadata must be an object") if replay_directory not in ("Replay", "Replays"): raise ValueError("unsupported replay directory") replay_sha1, replay_size = sha1_file(replay, MAX_REPLAY_BYTES) expected_replay = metadata.get("replaySha1") if expected_replay is not None and expected_replay != replay_sha1: raise ValueError("replay identity differs from metadata") replay_target = (prefix / "drive_c/users/ubuntu/Documents/Warcraft III" / replay_directory / "trace.w3g") install(replay, replay_target, replay_sha1, MAX_REPLAY_BYTES) result: dict[str, object] = { "replaySha1": replay_sha1, "replayBytes": replay_size, "mapInstalled": False, } if metadata.get("mapPath") is None: return result allow_expired = metadata.get("allowExpiredPatchMap") is True if allow_expired and metadata.get("engineProfile") != "native-1140": raise ValueError("expired patch maps require the native 1.14 profile") parts = map_parts(metadata.get("mapPath"), allow_expired) expected_map = metadata.get("mapContentSha1") or metadata.get("mapSha1") map_file = metadata.get("mapFile") if map_file is None and not metadata.get("mapRequired", False): result["mapPath"] = "\\".join(parts) return result if (not isinstance(expected_map, str) or len(expected_map) != 40 or any(character not in "0123456789abcdef" for character in expected_map) or not isinstance(map_file, str)): raise ValueError("map metadata is incomplete") source = safe_child(replay_root, map_file) actual_map, map_size = sha1_file(source, MAX_MAP_BYTES) if actual_map != expected_map: raise ValueError("map identity differs from metadata") documents = (prefix / "drive_c/users/ubuntu/Documents/Warcraft III").resolve() basename = parts[-1] destinations = { documents.joinpath(*parts), documents / "Maps" / basename, documents / "Maps" / "download" / basename, game_directory.joinpath(*parts), } for destination in destinations: parent = destination.parent.resolve() if os.path.commonpath((str(documents), str(parent))) == str(documents): allowed = documents elif os.path.commonpath((str(game_directory), str(parent))) == str(game_directory): allowed = game_directory else: raise ValueError("map destination escapes its session root") install(source, destination, expected_map, MAX_MAP_BYTES) result.update({ "mapInstalled": True, "mapPath": "\\".join(parts), "mapContentSha1": expected_map, "mapBytes": map_size, }) return result def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--metadata", required=True, type=Path) parser.add_argument("--replay", required=True, type=Path) parser.add_argument("--prefix", required=True, type=Path) parser.add_argument("--game-directory", required=True, type=Path) parser.add_argument("--replay-directory", choices=("Replay", "Replays"), default="Replays") args = parser.parse_args() print(json.dumps(stage(args.metadata, args.replay, args.prefix, args.game_directory, args.replay_directory), sort_keys=True)) if __name__ == "__main__": main()