from __future__ import annotations from dataclasses import dataclass from pathlib import Path import hashlib import gzip import shutil import tarfile import tomllib import zipfile VALID_SIDES = frozenset({"client", "server", "both"}) @dataclass(frozen=True) class PathRule: source: str side: str client_destination: str | None = None server_destination: str | None = None def load_side_manifest(path: Path) -> dict[str, str]: data = tomllib.loads(path.read_text(encoding="utf-8")) mapping = data["classification"] if not isinstance(mapping, dict): raise ValueError(f"{path}: [classification] is required") result = {str(name): str(side) for name, side in mapping.items()} invalid = sorted(name for name, side in result.items() if side not in VALID_SIDES) if invalid: raise ValueError(f"{path}: invalid side for {', '.join(invalid)}") return result def load_path_rules(path: Path) -> list[PathRule]: rules = [] for item in tomllib.loads(path.read_text(encoding="utf-8")).get("path", []): rule = PathRule( source=str(item["source"]), side=str(item["side"]), client_destination=item.get("client_destination"), server_destination=item.get("server_destination"), ) if rule.side not in VALID_SIDES: raise ValueError(f"{path}: invalid side for {rule.source}") if rule.side in {"client", "both"} and rule.client_destination is None: raise ValueError(f"{path}: missing client destination for {rule.source}") if rule.side in {"server", "both"} and rule.server_destination is None: raise ValueError(f"{path}: missing server destination for {rule.source}") rules.append(rule) return rules def validate_complete_mapping(expected: set[str], mapping: dict[str, str]) -> None: missing = sorted(expected - mapping.keys()) extra = sorted(mapping.keys() - expected) unknown = sorted(name for name, side in mapping.items() if side not in VALID_SIDES) if missing or extra or unknown: raise ValueError( f"missing={','.join(missing)} extra={','.join(extra)} unknown={','.join(unknown)}" ) def write_keybind_options(source: Path, destination: Path) -> None: kept = [ line for line in source.read_text(encoding="utf-8").splitlines() if line.startswith("version:") or line.startswith("key_") ] if sum(line.startswith("version:") for line in kept) != 1: raise ValueError(f"{source}: expected exactly one version line") destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text("\n".join(kept) + "\n", encoding="utf-8") def _matches_side(item_side: str, destination_side: str) -> bool: return item_side == "both" or item_side == destination_side def _safe_relative(value: str) -> Path: candidate = Path(value) if candidate.is_absolute() or ".." in candidate.parts: raise ValueError(f"unsafe relative path: {value}") return candidate def copy_classified_tree( source_root: Path, destination_root: Path, side: str, mod_map: dict[str, str], path_rules: list[PathRule], ) -> None: if side not in {"client", "server"}: raise ValueError(f"invalid destination side: {side}") mods_root = Path("minecraft/mods") if side == "client" else Path("mods") for jar_name, jar_side in mod_map.items(): if not _matches_side(jar_side, side): continue source = source_root / "minecraft/mods" / jar_name if not source.is_file(): raise FileNotFoundError(source) target = destination_root / mods_root / jar_name target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, target) for rule in path_rules: if not _matches_side(rule.side, side): continue destination_name = ( rule.client_destination if side == "client" else rule.server_destination ) if destination_name is None: raise ValueError(f"missing {side} destination for {rule.source}") source = source_root / _safe_relative(rule.source) if not source.exists(): raise FileNotFoundError(source) target = destination_root / _safe_relative(destination_name) if source.is_dir(): shutil.copytree(source, target, dirs_exist_ok=False) else: target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(source, target) CLIENT_FORBIDDEN_PREFIXES = ( "minecraft/saves/", "minecraft/logs/", "minecraft/crash-reports/", "minecraft/screenshots/", "minecraft/mods-disabled/", "minecraft/shaderpacks/", ) CLIENT_FORBIDDEN_NAMES = { "minecraft/servers.dat", "minecraft/optionsof.txt", "minecraft/optionsshaders.txt", } CLIENT_SANITIZED_PATHS = ( "minecraft/config/.backups", "minecraft/config/Discord-Integration.toml", "minecraft/config/EssentialsCommands-config.toml", "minecraft/config/NoChatReports/NCR-ServerPreferences.json", "minecraft/config/arctics_essentials", "minecraft/config/bstats.json", "minecraft/config/camerapture.server.json", "minecraft/config/chunky", "minecraft/config/discord_chat_mod-client.toml", "minecraft/config/discord_chat_mod-common.toml", "minecraft/config/dyairdrop.toml", "minecraft/config/elite-holograms", "minecraft/config/fancymenu/user_variables.db", "minecraft/config/maplink/friends.json5", "minecraft/config/oculus.properties", "minecraft/config/refurbished_furniture.server.toml", "minecraft/config/skinrestorer/config.json", "minecraft/config/skinrestorer/mojang_profile_cache.json", "minecraft/config/spark", "minecraft/config/voicechat/player-volumes.properties", "minecraft/config/voicechat/username-cache.json", "minecraft/config/voicechat/voicechat-server.properties", "minecraft/config/voicechat/voicechat-volumes.properties", "minecraft/config/worldedit", "minecraft/config/worldprotect-common.toml", ) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as input_file: while chunk := input_file.read(1024 * 1024): digest.update(chunk) return digest.hexdigest() def sanitize_client_stage(stage: Path) -> None: for relative in CLIENT_SANITIZED_PATHS: target = stage / relative if target.is_dir(): shutil.rmtree(target) elif target.is_file() or target.is_symlink(): target.unlink() def validate_client_stage(stage: Path, mod_map: dict[str, str]) -> None: for required in ("instance.cfg", "mmc-pack.json", "minecraft/options.txt"): if not (stage / required).is_file(): raise ValueError(f"missing client file: {required}") for path in sorted(stage.rglob("*")): relative = path.relative_to(stage).as_posix() if relative in CLIENT_FORBIDDEN_NAMES or relative.startswith(CLIENT_FORBIDDEN_PREFIXES): raise ValueError(f"forbidden client path: {relative}") for relative in CLIENT_SANITIZED_PATHS: if (stage / relative).exists(): raise ValueError(f"unsanitized client path: {relative}") actual = {path.name for path in (stage / "minecraft/mods").glob("*.jar")} expected = {name for name, side in mod_map.items() if side in {"client", "both"}} if actual != expected: raise ValueError( f"client JAR mismatch missing={sorted(expected - actual)} extra={sorted(actual - expected)}" ) def create_zip(source: Path, archive: Path) -> str: archive.parent.mkdir(parents=True, exist_ok=True) with zipfile.ZipFile(archive, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as output: for path in sorted(source.rglob("*")): if path.is_file(): info = zipfile.ZipInfo( path.relative_to(source).as_posix(), date_time=(2026, 7, 18, 0, 0, 0), ) info.compress_type = zipfile.ZIP_DEFLATED output.writestr(info, path.read_bytes()) return sha256_file(archive) def validate_server_stage(stage: Path, mod_map: dict[str, str], world_name: str) -> None: for required in ("eula.txt", "server.properties", "mods"): if not (stage / required).exists(): raise ValueError(f"missing server file: {required}") if not (stage / world_name).is_dir(): raise ValueError(f"missing target world: {world_name}") if (stage / "eula.txt").read_text(encoding="utf-8") != "eula=false\n": raise ValueError("server eula.txt must remain eula=false") properties = (stage / "server.properties").read_text(encoding="utf-8") if f"level-name={world_name}\n" not in properties: raise ValueError("server.properties does not target the expected world") actual = {path.name for path in (stage / "mods").glob("*.jar")} expected = {name for name, side in mod_map.items() if side in {"server", "both"}} if actual != expected: raise ValueError( f"server JAR mismatch missing={sorted(expected - actual)} extra={sorted(actual - expected)}" ) def create_tar_gz(source: Path, archive: Path) -> str: archive.parent.mkdir(parents=True, exist_ok=True) with archive.open("wb") as raw: with gzip.GzipFile(fileobj=raw, mode="wb", mtime=0) as compressed: with tarfile.open(fileobj=compressed, mode="w") as output: for path in sorted(source.rglob("*")): relative = path.relative_to(source).as_posix() info = output.gettarinfo(str(path), arcname=relative) info.uid = info.gid = 0 info.uname = info.gname = "" info.mtime = 0 if path.is_file(): with path.open("rb") as input_file: output.addfile(info, input_file) else: output.addfile(info) return sha256_file(archive)