"""Shared implementation for deterministic release building and verification.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone import fnmatch import gzip import hashlib import json import os from pathlib import Path, PurePosixPath import re import shutil import stat import subprocess import sys import tarfile import tempfile from typing import BinaryIO, Mapping, Sequence from urllib.parse import urlsplit PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) CONFIG_PATH = PROJECT_ROOT / "release/config.json" READ_SIZE = 256 * 1024 MAX_ARCHIVE_BYTES = 32 * 1024 * 1024 MAX_EXTRACTED_BYTES = 64 * 1024 * 1024 MAX_ARCHIVE_MEMBERS = 512 MAX_EMBEDDED_RELEASE_NOTES_CHARACTERS = 16_000 _ALLOWED_ARCHIVE_ROOTS = {"ptlinux", "ptdiag", "release.json"} _ARCHIVE_LAYOUTS = {"flat", "wrapped"} _SHA256_PATTERN = re.compile(r"[0-9a-f]{64}\Z") _COMMIT_PATTERN = re.compile(r"[0-9a-f]{7,64}\Z") _SOURCE_VERSION_PATTERN = re.compile( r'^__version__ = "([0-9]+\.[0-9]+\.[0-9]+(?:-(?:beta|rc)\.[0-9]+)?)"$', re.MULTILINE, ) _PROJECT_VERSION_PATTERN = re.compile( r'^version = "([0-9]+\.[0-9]+\.[0-9]+(?:-(?:beta|rc)\.[0-9]+)?)"$', re.MULTILINE, ) class ReleaseToolError(RuntimeError): """Raised when release artifacts cannot be built or trusted.""" @dataclass(frozen=True) class PayloadFile: relative_path: PurePosixPath source: Path | None data: bytes | None mode: int def load_config(path: Path = CONFIG_PATH) -> dict[str, object]: try: value = json.loads( path.read_text(encoding="utf-8"), object_pairs_hook=_unique_json_object, ) except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: raise ReleaseToolError(f"invalid release configuration {path}: {exc}") from exc if not isinstance(value, dict): raise ReleaseToolError("release configuration must be a JSON object") for key in ( "application", "archive_layout", "archive_prefix", "channels", "config_schema", "legacy_archive_root_prefix", "legacy_minimum_updater_version", "manifest_version", "minimum_config_schema", "minimum_updater_version", "release_branch", "release_status_file", "remote", "repository_url", "version_source", "wrapped_bridge_versions", "payload", "required_files", "exclusions", ): if key not in value: raise ReleaseToolError(f"release configuration is missing {key!r}") if value["archive_layout"] not in _ARCHIVE_LAYOUTS: raise ReleaseToolError( "release configuration archive_layout must be 'flat' or 'wrapped'" ) _string_list(value["wrapped_bridge_versions"], "wrapped_bridge_versions") return value def _archive_layout_for_version( config: Mapping[str, object], version: str, override: str | None = None, ) -> str: if override is not None: layout = override elif version in _string_list( config.get("wrapped_bridge_versions"), "wrapped_bridge_versions", ): layout = "wrapped" else: layout = _required_string(config, "archive_layout") if layout not in _ARCHIVE_LAYOUTS: raise ReleaseToolError(f"unsupported release archive layout: {layout!r}") return layout def _minimum_updater_for_layout( config: Mapping[str, object], archive_layout: str, ) -> str: key = ( "legacy_minimum_updater_version" if archive_layout == "wrapped" else "minimum_updater_version" ) return _required_string(config, key) def source_version(config: Mapping[str, object]) -> str: path = PROJECT_ROOT / _required_string(config, "version_source") try: contents = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise ReleaseToolError(f"cannot read launcher version from {path}: {exc}") from exc match = _SOURCE_VERSION_PATTERN.search(contents) if match is None: raise ReleaseToolError(f"cannot find ptlinux.__version__ in {path}") return match.group(1) def validate_version_mirrors(config: Mapping[str, object], expected: str) -> None: mirrors = _string_list(config.get("version_mirrors"), "version_mirrors") for relative in mirrors: path = PROJECT_ROOT / relative try: contents = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise ReleaseToolError(f"cannot read version mirror {path}: {exc}") from exc match = _PROJECT_VERSION_PATTERN.search(contents) if match is None or match.group(1) != expected: actual = match.group(1) if match is not None else "missing" raise ReleaseToolError( f"version mirror {relative} is {actual}, expected {expected}" ) def build_release( version: str, channel: str, output_directory: Path, *, repository_url: str | None = None, tag: str | None = None, commit: str | None = None, source_date_epoch: int | None = None, release_notes_file: Path | None = None, signing_key: Path | None = None, require_signature: bool = False, allow_source_version_override: bool = False, archive_layout: str | None = None, ) -> tuple[Path, ...]: from ptlinux.update_resolver import validate_channel_version config = load_config() parsed = validate_channel_version(channel, version) layout = _archive_layout_for_version(config, version, archive_layout) minimum_updater_version = _minimum_updater_for_layout(config, layout) expected_tag = parsed.tag if tag is not None and tag != expected_tag: raise ReleaseToolError(f"tag {tag!r} does not match version {version}") tag = expected_tag current_version = source_version(config) validate_version_mirrors(config, current_version) if current_version != version and not allow_source_version_override: raise ReleaseToolError( f"requested version {version} does not match launcher source {current_version}" ) commit = commit or _git_output("rev-parse", "HEAD") if _COMMIT_PATTERN.fullmatch(commit) is None: raise ReleaseToolError(f"invalid release commit: {commit!r}") epoch = source_date_epoch if epoch is None: epoch_text = os.environ.get("SOURCE_DATE_EPOCH") epoch = int(epoch_text) if epoch_text is not None else int( _git_output("show", "-s", "--format=%ct", commit) ) if epoch < 0: raise ReleaseToolError("SOURCE_DATE_EPOCH cannot be negative") repository = repository_url or _required_string(config, "repository_url") _validate_repository_url(repository) release_notes = _read_release_notes(release_notes_file) output = _safe_output_directory(output_directory) if signing_key is not None and signing_key.expanduser().resolve().is_relative_to(output): raise ReleaseToolError("the private signing key cannot be stored in release output") _clean_output_directory(output) archive_prefix = _required_string(config, "archive_prefix") archive_name = f"{archive_prefix}-{version}.tar.gz" archive = output / archive_name with tempfile.TemporaryDirectory(prefix="ptlinux-release-build-") as temporary: staging = Path(temporary) / "release" stage_release( staging, config, version=version, channel=channel, tag=tag, commit=commit, archive_layout=layout, ) if current_version != version: _set_staged_source_version(staging, config, version) _write_deterministic_archive( archive, staging, epoch, archive_layout=layout, archive_root=( f"{_required_string(config, 'legacy_archive_root_prefix')}{version}" if layout == "wrapped" else None ), ) checksum = archive.with_suffix(archive.suffix + ".sha256") _write_atomic( checksum, f"{sha256_file(archive)} {archive.name}\n".encode("ascii"), ) manifest = output / "manifest.json" generate_manifest( manifest, config=config, file_root=staging, archive=archive, version=version, channel=channel, tag=tag, commit=commit, repository_url=repository, release_timestamp=_timestamp_from_epoch(epoch), release_notes=release_notes, archive_layout=layout, minimum_updater_version=minimum_updater_version, ) artifacts: list[Path] = [archive, checksum, manifest] if release_notes_file is not None: release_notes_output = output / "release-notes.md" _write_atomic(release_notes_output, release_notes.encode("utf-8")) artifacts.append(release_notes_output) signature = output / "manifest.json.sig" if signing_key is not None: sign_file(manifest, signature, signing_key) verify_signature( manifest, signature, PROJECT_ROOT / "ptlinux/launcher-update-public.pem", ) artifacts.append(signature) elif require_signature: raise ReleaseToolError("a manifest signing key is required") return tuple(artifacts) def stage_release( destination: Path, config: Mapping[str, object], *, version: str, channel: str, tag: str, commit: str, archive_layout: str, ) -> None: if destination.exists(): raise ReleaseToolError(f"staging destination already exists: {destination}") destination.mkdir(parents=True, mode=0o755) payload = collect_payload(config) config_schema, minimum_config_schema = _config_schema_range(config) release_metadata = ( json.dumps( { "channel": channel, "commit": commit, "archive_layout": archive_layout, "config_schema": config_schema, "minimum_config_schema": minimum_config_schema, "schema": 1, "tag": tag, "version": version, }, indent=2, sort_keys=True, ) + "\n" ).encode("utf-8") payload.append( PayloadFile(PurePosixPath("release.json"), None, release_metadata, 0o644) ) seen: set[PurePosixPath] = set() for item in sorted(payload, key=lambda entry: entry.relative_path.as_posix()): if item.relative_path in seen: raise ReleaseToolError(f"duplicate payload path: {item.relative_path}") seen.add(item.relative_path) target = destination.joinpath(*item.relative_path.parts) target.parent.mkdir(parents=True, exist_ok=True, mode=0o755) if item.source is not None: data = item.source.read_bytes() else: assert item.data is not None data = item.data target.write_bytes(data) target.chmod(item.mode) for directory in (destination, *destination.rglob("*")): if directory.is_dir(): directory.chmod(0o755) for required in _string_list(config.get("required_files"), "required_files"): if not (destination / required).is_file(): raise ReleaseToolError(f"staging tree is missing required file: {required}") def collect_payload(config: Mapping[str, object]) -> list[PayloadFile]: payload_config = config.get("payload") if not isinstance(payload_config, dict): raise ReleaseToolError("release payload configuration must be an object") result: list[PayloadFile] = [] python_trees = payload_config.get("python_trees") if not isinstance(python_trees, list): raise ReleaseToolError("payload.python_trees must be an array") for tree in python_trees: if not isinstance(tree, dict): raise ReleaseToolError("payload python tree must be an object") source_name = _required_string(tree, "source") source_root = PROJECT_ROOT / source_name if not source_root.is_dir() or source_root.is_symlink(): raise ReleaseToolError(f"payload source tree is invalid: {source_name}") exclusions = set(_string_list(tree.get("exclude", []), "payload exclude")) for source in sorted(source_root.rglob("*.py")): relative_in_tree = source.relative_to(source_root).as_posix() if relative_in_tree in exclusions: continue _validate_source_file(source) result.append( PayloadFile( PurePosixPath(source_name) / PurePosixPath(relative_in_tree), source, None, 0o644, ) ) configured_files = payload_config.get("files") if not isinstance(configured_files, list): raise ReleaseToolError("payload.files must be an array") for file_config in configured_files: if not isinstance(file_config, dict): raise ReleaseToolError("payload file entry must be an object") path_text = _required_string(file_config, "path") relative = _safe_relative_path(path_text) source = PROJECT_ROOT.joinpath(*relative.parts) _validate_source_file(source) mode_text = _required_string(file_config, "mode") try: mode = int(mode_text, 8) except ValueError as exc: raise ReleaseToolError(f"invalid payload mode {mode_text!r}") from exc if mode not in {0o644, 0o755}: raise ReleaseToolError(f"unsupported payload mode: {mode_text}") result.append(PayloadFile(relative, source, None, mode)) return result def generate_manifest( destination: Path, *, config: Mapping[str, object], file_root: Path, archive: Path, version: str, channel: str, tag: str, commit: str, repository_url: str, release_timestamp: str, release_notes: str = "", release_title: str | None = None, release_notes_url: str | None = None, archive_layout: str | None = None, minimum_updater_version: str | None = None, ) -> dict[str, object]: from ptlinux.update_resolver import validate_channel_version parsed = validate_channel_version(channel, version) layout = _archive_layout_for_version(config, version, archive_layout) if tag != parsed.tag: raise ReleaseToolError(f"tag {tag!r} does not match version {version}") if _COMMIT_PATTERN.fullmatch(commit) is None: raise ReleaseToolError(f"invalid release commit: {commit!r}") if not file_root.is_dir() or file_root.is_symlink(): raise ReleaseToolError(f"manifest file root is invalid: {file_root}") if not archive.is_file() or archive.is_symlink(): raise ReleaseToolError(f"release archive is missing: {archive}") _validate_repository_url(repository_url) _parse_timestamp(release_timestamp) repository = repository_url.rstrip("/") digest = sha256_file(archive) # Gitea release URLs exist before their attachments do. A CDN can cache that # first empty response under the otherwise immutable asset URL, then keep # serving it after the real archive is uploaded. The signed content digest # gives the completed asset its own cache key without weakening verification. archive_url = ( f"{repository}/releases/download/{tag}/{archive.name}?sha256={digest}" ) notes_url = release_notes_url or f"{repository}/releases/tag/{tag}" files: list[dict[str, object]] = [] for path in sorted(file_root.rglob("*"), key=lambda item: item.relative_to(file_root).as_posix()): if path.is_symlink(): raise ReleaseToolError(f"staging tree contains a symbolic link: {path}") if not path.is_file(): continue relative = path.relative_to(file_root).as_posix() files.append( { "path": relative, "sha256": sha256_file(path), "size": path.stat().st_size, } ) size = archive.stat().st_size title = release_title or f"PristonTale EU Launcher {version}" summary = _release_summary(release_notes, version) embedded_release_notes = release_notes or f"PristonTale EU Launcher {version}.\n" if len(embedded_release_notes) > MAX_EMBEDDED_RELEASE_NOTES_CHARACTERS: embedded_release_notes = ( embedded_release_notes[:MAX_EMBEDDED_RELEASE_NOTES_CHARACTERS].rstrip() + "\n\nRelease notes continue at the signed release URL.\n" ) minimum = minimum_updater_version or _minimum_updater_for_layout(config, layout) manifest_schema = _required_integer(config, "manifest_version") config_schema, minimum_config_schema = _config_schema_range(config) manifest: dict[str, object] = { "application": _required_string(config, "application"), "archive": { "layout": layout, "name": archive.name, "sha256": digest, "size": size, "url": archive_url, }, # Legacy schema-1 fields remain during migration for deployed stable clients. "archive_sha256": digest, "archive_size": size, "archive_url": archive_url, "channel": channel, "commit": commit, "compatibility": { "config_schema": config_schema, "manifest_schema": manifest_schema, "minimum_config_schema": minimum_config_schema, "minimum_updater_version": minimum, }, "files": files, "minimum_updater_version": minimum, "notes_url": notes_url, "published_at": release_timestamp, "release": { "notes_markdown": embedded_release_notes, "release_notes_url": notes_url, "summary": summary, "title": title, }, "schema": manifest_schema, "status": "active", "tag": tag, "version": version, } _write_atomic( destination, (json.dumps(manifest, indent=2, sort_keys=True) + "\n").encode("utf-8"), ) return manifest def verify_release( version: str, channel: str, output_directory: Path, *, public_key: Path | None = None, require_signature: bool = False, ) -> None: from ptlinux.update_resolver import validate_channel_version config = load_config() parsed = validate_channel_version(channel, version) output = output_directory.expanduser().resolve() archive = output / f"{_required_string(config, 'archive_prefix')}-{version}.tar.gz" checksum = archive.with_suffix(archive.suffix + ".sha256") manifest_path = output / "manifest.json" for required in (archive, checksum, manifest_path): if not required.is_file() or required.is_symlink(): raise ReleaseToolError(f"required release artifact is missing: {required}") if archive.stat().st_size <= 0 or archive.stat().st_size > MAX_ARCHIVE_BYTES: raise ReleaseToolError("release archive has an invalid size") checksum_text = checksum.read_text(encoding="ascii") expected_checksum = f"{sha256_file(archive)} {archive.name}\n" if checksum_text != expected_checksum: raise ReleaseToolError("archive checksum file does not match the archive") manifest = _read_json_object(manifest_path) archive_layout = _validate_manifest_artifact( manifest, config=config, version=version, channel=channel, tag=parsed.tag, archive=archive, ) manifest_files = _manifest_files(manifest) with tempfile.TemporaryDirectory(prefix="ptlinux-release-verify-") as temporary: extracted = Path(temporary) / "release" _extract_verified_archive( archive, extracted, archive_layout=archive_layout, archive_root=( f"{_required_string(config, 'legacy_archive_root_prefix')}{version}" if archive_layout == "wrapped" else None ), manifest_files=manifest_files, config=config, ) for required in _string_list(config.get("required_files"), "required_files"): if not (extracted / required).is_file(): raise ReleaseToolError(f"release archive is missing required file: {required}") packaged_version = _version_from_source_file( extracted / _required_string(config, "version_source") ) if packaged_version != version: raise ReleaseToolError( f"packaged launcher version is {packaged_version}, expected {version}" ) release_metadata = _read_json_object(extracted / "release.json") for key, expected in ( ("version", version), ("channel", channel), ("tag", parsed.tag), ("archive_layout", archive_layout), ): if release_metadata.get(key) != expected: raise ReleaseToolError( f"release.json {key} does not match the release: " f"{release_metadata.get(key)!r}" ) signature = output / "manifest.json.sig" if signature.exists(): if not signature.is_file() or signature.is_symlink(): raise ReleaseToolError("manifest signature is not a regular file") verify_signature( manifest_path, signature, public_key or PROJECT_ROOT / "ptlinux/launcher-update-public.pem", ) elif require_signature: raise ReleaseToolError("manifest signature is required but missing") def sign_file(source: Path, signature: Path, private_key: Path) -> None: key = private_key.expanduser().resolve() if not key.is_file() or key.is_symlink(): raise ReleaseToolError(f"manifest signing key is missing: {key}") openssl = shutil.which("openssl") if openssl is None: raise ReleaseToolError("OpenSSL is required to sign release manifests") temporary = signature.with_name(f".{signature.name}.{os.getpid()}.tmp") temporary.unlink(missing_ok=True) commands = ( [openssl, "dgst", "-sha256", "-sign", str(key), "-out", str(temporary), str(source)], [openssl, "pkeyutl", "-sign", "-rawin", "-inkey", str(key), "-in", str(source), "-out", str(temporary)], ) last_error = "" try: for command in commands: temporary.unlink(missing_ok=True) result = subprocess.run( command, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30, check=False, ) if result.returncode == 0 and temporary.is_file(): temporary.chmod(0o644) os.replace(temporary, signature) return last_error = result.stderr.strip() except (OSError, subprocess.TimeoutExpired) as exc: last_error = str(exc) finally: temporary.unlink(missing_ok=True) raise ReleaseToolError(f"OpenSSL could not sign manifest: {last_error}") def verify_signature(source: Path, signature: Path, public_key: Path) -> None: key = public_key.expanduser().resolve() if not key.is_file() or key.is_symlink(): raise ReleaseToolError(f"manifest public key is missing: {key}") openssl = shutil.which("openssl") if openssl is None: raise ReleaseToolError("OpenSSL is required to verify release manifests") commands = ( [ openssl, "dgst", "-sha256", "-verify", str(key), "-signature", str(signature), str(source), ], [ openssl, "pkeyutl", "-verify", "-rawin", "-pubin", "-inkey", str(key), "-sigfile", str(signature), "-in", str(source), ], ) for command in commands: try: result = subprocess.run( command, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=20, check=False, ) except (OSError, subprocess.TimeoutExpired): continue if result.returncode == 0: return raise ReleaseToolError("manifest signature verification failed") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(READ_SIZE), b""): digest.update(chunk) return digest.hexdigest() def _write_deterministic_archive( destination: Path, staging: Path, mtime: int, *, archive_layout: str, archive_root: str | None, ) -> None: if archive_layout not in _ARCHIVE_LAYOUTS: raise ReleaseToolError(f"unsupported release archive layout: {archive_layout!r}") if archive_layout == "wrapped" and not archive_root: raise ReleaseToolError("wrapped release archive requires a root directory") if archive_layout == "flat" and archive_root is not None: raise ReleaseToolError("flat release archive cannot have a root directory") temporary = destination.with_name(f".{destination.name}.{os.getpid()}.tmp") temporary.unlink(missing_ok=True) try: with temporary.open("xb") as raw: with gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=mtime) as compressed: with tarfile.open(fileobj=compressed, mode="w", format=tarfile.PAX_FORMAT) as bundle: if archive_root is not None: information = _tar_info(archive_root, mtime, 0o755) information.type = b"5" bundle.addfile(information) paths = list(staging.rglob("*")) paths.sort(key=lambda item: item.relative_to(staging).as_posix()) for path in paths: relative = path.relative_to(staging).as_posix() archive_name = ( f"{archive_root}/{relative}" if archive_root is not None else relative ) if path.is_symlink(): raise ReleaseToolError(f"staging tree contains a symlink: {path}") if path.is_dir(): information = _tar_info(archive_name, mtime, 0o755) information.type = b"5" bundle.addfile(information) elif path.is_file(): mode = stat.S_IMODE(path.stat().st_mode) with path.open("rb") as handle: _add_tar_file( bundle, information_name=archive_name, handle=handle, size=path.stat().st_size, mtime=mtime, mode=mode, ) else: raise ReleaseToolError( f"staging tree contains a special file: {path}" ) temporary.chmod(0o644) os.replace(temporary, destination) finally: temporary.unlink(missing_ok=True) def _tar_info(name: str, mtime: int, mode: int) -> tarfile.TarInfo: information = tarfile.TarInfo(name) information.uid = 0 information.gid = 0 information.uname = "" information.gname = "" information.mtime = mtime information.mode = mode return information def _add_tar_file( bundle: tarfile.TarFile, *, information_name: str, handle: BinaryIO, size: int, mtime: int, mode: int, ) -> None: information = _tar_info(information_name, mtime, mode) information.size = size bundle.addfile(information, handle) def _validate_manifest_artifact( manifest: Mapping[str, object], *, config: Mapping[str, object], version: str, channel: str, tag: str, archive: Path, ) -> str: expected_archive_path = f"/{tag}/{archive.name}" expected = { "schema": _required_integer(config, "manifest_version"), "application": _required_string(config, "application"), "version": version, "channel": channel, "tag": tag, "status": "active", } for key in ("schema", "application", "version", "channel", "tag", "status"): if manifest.get(key) != expected[key]: raise ReleaseToolError( f"manifest {key} is {manifest.get(key)!r}, expected {expected[key]!r}" ) commit = manifest.get("commit") if not isinstance(commit, str) or _COMMIT_PATTERN.fullmatch(commit) is None: raise ReleaseToolError("manifest commit is invalid") published_at = manifest.get("published_at") if not isinstance(published_at, str): raise ReleaseToolError("manifest published_at is invalid") _parse_timestamp(published_at) digest = sha256_file(archive) size = archive.stat().st_size if manifest.get("archive_sha256") != digest or manifest.get("archive_size") != size: raise ReleaseToolError("legacy manifest archive metadata does not match") archive_url = manifest.get("archive_url") archive_parts = urlsplit(archive_url) if isinstance(archive_url, str) else None if ( archive_parts is None or archive_parts.scheme.casefold() != "https" or not archive_parts.netloc or archive_parts.path.endswith(expected_archive_path) is False or archive_parts.query not in {"", f"sha256={digest}"} or archive_parts.fragment ): raise ReleaseToolError("manifest archive URL does not match tag and filename") archive_value = manifest.get("archive") if not isinstance(archive_value, dict): raise ReleaseToolError("manifest archive metadata must be an object") archive_layout = archive_value.get("layout", "wrapped") if archive_layout not in _ARCHIVE_LAYOUTS: raise ReleaseToolError("manifest archive layout is invalid") if ( archive_value.get("name") != archive.name or archive_value.get("size") != size or archive_value.get("sha256") != digest or archive_value.get("url") != archive_url ): raise ReleaseToolError("manifest archive object does not match the archive") expected_minimum = _minimum_updater_for_layout(config, archive_layout) if manifest.get("minimum_updater_version") != expected_minimum: raise ReleaseToolError("manifest minimum updater does not match archive layout") compatibility = manifest.get("compatibility") if ( not isinstance(compatibility, dict) or compatibility.get("minimum_updater_version") != expected_minimum ): raise ReleaseToolError( "manifest compatibility minimum updater does not match archive layout" ) release = manifest.get("release") if not isinstance(release, dict): raise ReleaseToolError("manifest release metadata must be an object") for key in ("title", "summary", "release_notes_url", "notes_markdown"): if not isinstance(release.get(key), str) or not release[key]: raise ReleaseToolError(f"manifest release.{key} must be a non-empty string") return archive_layout def _manifest_files(manifest: Mapping[str, object]) -> dict[str, tuple[int, str]]: value = manifest.get("files") if ( not isinstance(value, list) or not value or len(value) > MAX_ARCHIVE_MEMBERS ): raise ReleaseToolError("manifest files must be a non-empty array") result: dict[str, tuple[int, str]] = {} last_path = "" for entry in value: if not isinstance(entry, dict): raise ReleaseToolError("manifest file entry must be an object") path = entry.get("path") size = entry.get("size") digest = entry.get("sha256") if not isinstance(path, str): raise ReleaseToolError("manifest file path must be a string") _safe_relative_path(path) if path <= last_path or path in result: raise ReleaseToolError("manifest files must be uniquely sorted by path") if isinstance(size, bool) or not isinstance(size, int) or size < 0: raise ReleaseToolError(f"manifest file size is invalid: {path}") if not isinstance(digest, str) or _SHA256_PATTERN.fullmatch(digest) is None: raise ReleaseToolError(f"manifest file SHA-256 is invalid: {path}") result[path] = size, digest last_path = path return result def _extract_verified_archive( archive: Path, destination: Path, *, archive_layout: str, archive_root: str | None, manifest_files: Mapping[str, tuple[int, str]], config: Mapping[str, object], ) -> None: if archive_layout not in _ARCHIVE_LAYOUTS: raise ReleaseToolError(f"unsupported release archive layout: {archive_layout!r}") if archive_layout == "wrapped" and not archive_root: raise ReleaseToolError("wrapped release archive requires a root directory") extracted_files: set[str] = set() seen_paths: set[PurePosixPath] = set() manifest_directories = { parent.as_posix() for name in manifest_files for parent in PurePosixPath(name).parents if parent.parts } total_size = 0 try: bundle = tarfile.open(archive, "r:gz") except (OSError, tarfile.TarError) as exc: raise ReleaseToolError(f"archive cannot be opened: {exc}") from exc with bundle: members = bundle.getmembers() if not members or len(members) > MAX_ARCHIVE_MEMBERS: raise ReleaseToolError("archive has an invalid member count") for member in members: name = ( member.name[:-1] if member.isdir() and member.name.endswith("/") else member.name ) raw_parts = name.split("/") if ( not name or "\\" in name or "\x00" in name or any(part in {"", ".", ".."} for part in raw_parts) ): raise ReleaseToolError(f"archive contains unsafe path: {member.name!r}") path = PurePosixPath(name) if path.is_absolute() or path in seen_paths: raise ReleaseToolError( f"archive contains unsafe or duplicate path: {member.name!r}" ) seen_paths.add(path) if archive_layout == "wrapped": if path.parts[0] != archive_root: raise ReleaseToolError( f"archive member is outside {archive_root}: {member.name}" ) relative = PurePosixPath(*path.parts[1:]) if not relative.parts: if not member.isdir(): raise ReleaseToolError("archive root must be a directory") root_mode = member.mode & 0o777 if root_mode != 0o755: raise ReleaseToolError("archive root has unsafe permissions") continue else: relative = path if member.issym() or member.islnk() or not (member.isdir() or member.isfile()): raise ReleaseToolError(f"archive contains unsupported member: {member.name}") if relative.parts[0] not in _ALLOWED_ARCHIVE_ROOTS: raise ReleaseToolError(f"archive contains unexpected root path: {relative}") if relative.parts[0] == "release.json" and ( len(relative.parts) != 1 or not member.isfile() ): raise ReleaseToolError( f"archive contains invalid release metadata path: {relative}" ) if ( len(relative.parts) == 1 and relative.parts[0] in {"ptlinux", "ptdiag"} and not member.isdir() ): raise ReleaseToolError( f"archive package root must be a directory: {relative}" ) mode = member.mode & 0o777 if mode & 0o022 or mode not in {0o644, 0o755}: raise ReleaseToolError(f"archive member has unsafe permissions: {member.name}") _forbid_release_path(relative, config) target = destination.joinpath(*relative.parts) if member.isdir(): if relative.as_posix() not in manifest_directories: raise ReleaseToolError( f"archive directory is absent from manifest paths: {relative}" ) target.mkdir(parents=True, exist_ok=True, mode=0o755) continue if relative.as_posix() in extracted_files: raise ReleaseToolError(f"archive contains duplicate path: {relative}") expected = manifest_files.get(relative.as_posix()) if expected is None: raise ReleaseToolError(f"archive file is absent from manifest: {relative}") if member.size != expected[0] or member.size < 0: raise ReleaseToolError(f"archive file size differs from manifest: {relative}") total_size += member.size if total_size > MAX_EXTRACTED_BYTES: raise ReleaseToolError("archive exceeds the extraction size limit") source = bundle.extractfile(member) if source is None: raise ReleaseToolError(f"archive file cannot be read: {relative}") data = source.read(member.size + 1) if len(data) != member.size: raise ReleaseToolError(f"archive file is truncated: {relative}") if hashlib.sha256(data).hexdigest() != expected[1]: raise ReleaseToolError(f"archive file hash differs from manifest: {relative}") target.parent.mkdir(parents=True, exist_ok=True, mode=0o755) target.write_bytes(data) target.chmod(mode) extracted_files.add(relative.as_posix()) if extracted_files != set(manifest_files): missing = sorted(set(manifest_files).difference(extracted_files)) raise ReleaseToolError(f"manifest files are absent from archive: {', '.join(missing)}") def _forbid_release_path(path: PurePosixPath, config: Mapping[str, object]) -> None: text = path.as_posix() allowed_pem = set(_string_list(config.get("allowed_pem_files", []), "allowed_pem_files")) if text in allowed_pem: return exclusions = _string_list(config.get("exclusions"), "exclusions") for pattern in exclusions: if ( fnmatch.fnmatch(text, pattern) or fnmatch.fnmatch(path.name, pattern) or pattern in path.parts ): raise ReleaseToolError(f"archive contains forbidden path: {text}") def _safe_output_directory(path: Path) -> Path: expanded = path.expanduser() if expanded.is_symlink(): raise ReleaseToolError(f"output directory cannot be a symbolic link: {expanded}") resolved = expanded.resolve() project_dist_path = PROJECT_ROOT / "dist" if project_dist_path.is_symlink(): raise ReleaseToolError( f"repository dist directory cannot be a symbolic link: {project_dist_path}" ) project_dist = project_dist_path.resolve() temporary_roots = ( Path(tempfile.gettempdir()).resolve(), Path("/var/tmp").resolve(), ) unsafe = { Path("/").resolve(), PROJECT_ROOT.resolve(), Path.home().resolve(), *temporary_roots, } if resolved in unsafe or resolved in PROJECT_ROOT.resolve().parents: raise ReleaseToolError(f"refusing unsafe output directory: {resolved}") if resolved == project_dist or resolved.is_relative_to(project_dist): return resolved if any(resolved.is_relative_to(root) for root in temporary_roots): return resolved raise ReleaseToolError( "release output must be the repository dist directory or a temporary directory" ) def _clean_output_directory(path: Path) -> None: path.mkdir(parents=True, exist_ok=True) for child in path.iterdir(): if child.is_symlink() or child.is_file(): child.unlink() elif child.is_dir(): shutil.rmtree(child) else: raise ReleaseToolError(f"cannot clean special output path: {child}") def _validate_source_file(path: Path) -> None: if path.is_symlink() or not path.is_file(): raise ReleaseToolError(f"required payload file is missing or unsafe: {path}") def _safe_relative_path(value: str) -> PurePosixPath: if "\\" in value or "\x00" in value: raise ReleaseToolError(f"unsafe relative release path: {value!r}") raw_parts = value.split("/") if any(part in {"", ".", ".."} for part in raw_parts): raise ReleaseToolError(f"unsafe relative release path: {value!r}") path = PurePosixPath(value) if path.is_absolute() or not path.parts: raise ReleaseToolError(f"unsafe relative release path: {value!r}") return path def _read_release_notes(path: Path | None) -> str: if path is None: return "" try: value = path.expanduser().resolve().read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise ReleaseToolError(f"cannot read release notes {path}: {exc}") from exc if not value.endswith("\n"): value += "\n" return value def _release_summary(notes: str, version: str) -> str: for line in notes.splitlines(): candidate = line.strip() if not candidate or candidate.startswith("#"): continue candidate = re.sub(r"^[-*+]\s+", "", candidate) return candidate[:280] return f"PristonTale EU Launcher {version}." def _timestamp_from_epoch(epoch: int) -> str: return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat().replace("+00:00", "Z") def _parse_timestamp(value: str) -> datetime: try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise ReleaseToolError(f"invalid release timestamp: {value!r}") from exc if parsed.tzinfo is None: raise ReleaseToolError("release timestamp must include a timezone") return parsed def _validate_repository_url(value: str) -> None: parts = urlsplit(value) if ( parts.scheme.casefold() != "https" or not parts.netloc or parts.username is not None or parts.password is not None or parts.query or parts.fragment ): raise ReleaseToolError( "repository URL must be HTTPS without credentials, query text, or a fragment" ) def _read_json_object(path: Path) -> dict[str, object]: try: value = json.loads( path.read_text(encoding="utf-8"), object_pairs_hook=_unique_json_object, ) except (OSError, UnicodeError, json.JSONDecodeError, ValueError) as exc: raise ReleaseToolError(f"invalid JSON file {path}: {exc}") from exc if not isinstance(value, dict): raise ReleaseToolError(f"JSON file must contain an object: {path}") return value def _required_string(value: Mapping[str, object], key: str) -> str: item = value.get(key) if not isinstance(item, str) or not item: raise ReleaseToolError(f"release configuration {key!r} must be a non-empty string") return item def _required_integer(value: Mapping[str, object], key: str) -> int: item = value.get(key) if isinstance(item, bool) or not isinstance(item, int) or item <= 0: raise ReleaseToolError( f"release configuration {key!r} must be a positive integer" ) return item def _config_schema_range(config: Mapping[str, object]) -> tuple[int, int]: config_schema = _required_integer(config, "config_schema") minimum_config_schema = _required_integer(config, "minimum_config_schema") if minimum_config_schema > config_schema: raise ReleaseToolError( "minimum_config_schema cannot exceed config_schema" ) return config_schema, minimum_config_schema def _version_from_source_file(path: Path) -> str: try: contents = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise ReleaseToolError(f"cannot read launcher version from {path}: {exc}") from exc match = _SOURCE_VERSION_PATTERN.search(contents) if match is None: raise ReleaseToolError(f"cannot find ptlinux.__version__ in {path}") return match.group(1) def _set_staged_source_version( staging: Path, config: Mapping[str, object], version: str, ) -> None: path = staging / _required_string(config, "version_source") try: contents = path.read_text(encoding="utf-8") except (OSError, UnicodeError) as exc: raise ReleaseToolError( f"cannot prepare staged launcher version in {path}: {exc}" ) from exc replacement = f'__version__ = "{version}"' updated, count = _SOURCE_VERSION_PATTERN.subn(replacement, contents) if count != 1: raise ReleaseToolError( f"cannot prepare staged launcher version in {path}: expected one version" ) _write_atomic(path, updated.encode("utf-8")) def _string_list(value: object, label: str) -> list[str]: if not isinstance(value, list) or not all(isinstance(item, str) and item for item in value): raise ReleaseToolError(f"{label} must be an array of non-empty strings") return list(value) def _unique_json_object(pairs: Sequence[tuple[str, object]]) -> dict[str, object]: value: dict[str, object] = {} for key, item in pairs: if key in value: raise ValueError(f"duplicate JSON key: {key!r}") value[key] = item return value def _write_atomic(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") temporary.unlink(missing_ok=True) try: with temporary.open("xb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) temporary.chmod(0o644) os.replace(temporary, path) finally: temporary.unlink(missing_ok=True) def _git_output(*arguments: str) -> str: try: result = subprocess.run( ["git", "-C", str(PROJECT_ROOT), *arguments], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise ReleaseToolError(f"cannot inspect Git release source: {exc}") from exc if result.returncode != 0: raise ReleaseToolError(result.stderr.strip() or "Git release source check failed") return result.stdout.strip()