From 8afd52ae57f9237f4f46138ad224dfc621d7ba59 Mon Sep 17 00:00:00 2001 From: Can Celik Date: Sat, 25 Jul 2026 02:40:14 +0300 Subject: fix: bundle modern ConPTY on Windows (#1828) refs #1533 refs #1644 --- scripts/package_windows_conpty.py | 245 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 245 insertions(+) create mode 100644 scripts/package_windows_conpty.py (limited to 'scripts/package_windows_conpty.py') diff --git a/scripts/package_windows_conpty.py b/scripts/package_windows_conpty.py new file mode 100644 index 00000000..2fc51a1e --- /dev/null +++ b/scripts/package_windows_conpty.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import hashlib +import json +import shutil +import struct +import tempfile +import urllib.request +import xml.etree.ElementTree as ET +import zipfile +from pathlib import Path, PurePosixPath +from typing import Any + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +DEFAULT_METADATA = PROJECT_ROOT / "packaging" / "windows" / "conpty.json" +MARKER_PATH = PurePosixPath("conpty/herdr-conpty.json") +DOWNLOAD_TIMEOUT_SECONDS = 60 + + +def sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def load_metadata(path: Path) -> dict[str, Any]: + data = json.loads(path.read_text(encoding="utf-8")) + if data.get("schema_version") != 1: + raise ValueError("unsupported ConPTY metadata schema") + return data + + +def pe_machine(data: bytes) -> int: + if len(data) < 0x40 or data[:2] != b"MZ": + raise ValueError("file is not a PE image") + pe_offset = struct.unpack_from(" len(data) or data[pe_offset : pe_offset + 4] != b"PE\0\0": + raise ValueError("file has an invalid PE header") + return struct.unpack_from(" None: + nuspec_name = f"{package['id']}.nuspec" + try: + root = ET.fromstring(archive.read(nuspec_name).decode("utf-8-sig")) + except KeyError as error: + raise ValueError(f"package is missing {nuspec_name}") from error + namespace = {"n": "http://schemas.microsoft.com/packaging/2011/08/nuspec.xsd"} + metadata = root.find("n:metadata", namespace) + if metadata is None: + raise ValueError("package NuSpec is missing metadata") + expected = { + "id": package["id"], + "version": package["version"], + "license": package["license"], + } + for field, expected_value in expected.items(): + element = metadata.find(f"n:{field}", namespace) + actual = element.text.strip() if element is not None and element.text else None + if actual != expected_value: + raise ValueError( + f"package NuSpec {field} mismatch: expected {expected_value!r}, got {actual!r}" + ) + + +def acquire_package(package: dict[str, Any], package_path: Path) -> None: + package_path.parent.mkdir(parents=True, exist_ok=True) + if not package_path.exists(): + with urllib.request.urlopen( + package["url"], timeout=DOWNLOAD_TIMEOUT_SECONDS + ) as response, package_path.open("wb") as output: + shutil.copyfileobj(response, output) + actual = sha256_file(package_path) + if actual != package["sha256"]: + raise ValueError( + f"ConPTY package hash mismatch: expected {package['sha256']}, got {actual}" + ) + + +def marker_data(metadata: dict[str, Any], architecture: str) -> bytes: + bundle = metadata["bundles"][architecture] + marker = { + "schema_version": 1, + "package": metadata["package"]["id"], + "version": metadata["package"]["version"], + "architecture": architecture, + "files": { + item["destination"]: item["sha256"] for item in bundle["files"] + }, + } + return (json.dumps(marker, indent=2, sort_keys=True) + "\n").encode() + + +def stage_bundle( + metadata_path: Path, + architecture: str, + package_path: Path, + herdr_exe: Path, + output_dir: Path, +) -> None: + metadata = load_metadata(metadata_path) + if architecture not in metadata["bundles"]: + raise ValueError(f"unsupported Windows architecture: {architecture}") + if output_dir.exists(): + raise ValueError(f"output directory already exists: {output_dir}") + if not herdr_exe.is_file(): + raise ValueError(f"Herdr executable does not exist: {herdr_exe}") + + acquire_package(metadata["package"], package_path) + bundle = metadata["bundles"][architecture] + metadata_root = metadata_path.resolve().parent + + with zipfile.ZipFile(package_path) as archive, tempfile.TemporaryDirectory( + prefix="herdr-conpty-stage-", dir=output_dir.parent + ) as temporary: + validate_nuspec(archive, metadata["package"]) + staging = Path(temporary) / "bundle" + staging.mkdir() + shutil.copy2(herdr_exe, staging / "herdr.exe") + + for item in bundle["files"]: + try: + payload = archive.read(item["source"]) + except KeyError as error: + raise ValueError(f"package is missing {item['source']}") from error + actual_hash = sha256_bytes(payload) + if actual_hash != item["sha256"]: + raise ValueError( + f"hash mismatch for {item['source']}: expected {item['sha256']}, got {actual_hash}" + ) + actual_machine = pe_machine(payload) + expected_machine = int(item["pe_machine"], 16) + if actual_machine != expected_machine: + raise ValueError( + f"PE machine mismatch for {item['source']}: " + f"expected {item['pe_machine']}, got 0x{actual_machine:04x}" + ) + destination = staging / PurePosixPath(item["destination"]) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(payload) + + marker = staging / MARKER_PATH + marker.parent.mkdir(parents=True, exist_ok=True) + marker.write_bytes(marker_data(metadata, architecture)) + + for notice in metadata["notices"]: + source = (metadata_root.parent.parent / notice["source"]).resolve() + actual_hash = sha256_file(source) + if actual_hash != notice["sha256"]: + raise ValueError( + f"notice hash mismatch for {source}: expected {notice['sha256']}, got {actual_hash}" + ) + destination = staging / PurePosixPath(notice["destination"]) + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + + staging.rename(output_dir) + + +def expected_stage_files(metadata: dict[str, Any], architecture: str) -> set[str]: + files = {"herdr.exe", MARKER_PATH.as_posix()} + files.update(item["destination"] for item in metadata["bundles"][architecture]["files"]) + files.update(item["destination"] for item in metadata["notices"]) + return files + + +def validate_stage(metadata_path: Path, architecture: str, stage_dir: Path) -> None: + metadata = load_metadata(metadata_path) + actual = { + path.relative_to(stage_dir).as_posix() + for path in stage_dir.rglob("*") + if path.is_file() + } + expected = expected_stage_files(metadata, architecture) + if actual != expected: + raise ValueError( + f"bundle layout mismatch; missing={sorted(expected - actual)}, unexpected={sorted(actual - expected)}" + ) + if (stage_dir / MARKER_PATH).read_bytes() != marker_data(metadata, architecture): + raise ValueError("bundle marker does not match pinned ConPTY metadata") + for item in metadata["bundles"][architecture]["files"]: + path = stage_dir / PurePosixPath(item["destination"]) + actual_hash = sha256_file(path) + if actual_hash != item["sha256"]: + raise ValueError(f"staged file hash mismatch for {path}") + for notice in metadata["notices"]: + path = stage_dir / PurePosixPath(notice["destination"]) + actual_hash = sha256_file(path) + if actual_hash != notice["sha256"]: + raise ValueError(f"staged notice hash mismatch for {path}") + + +def archive_bundle( + metadata_path: Path, architecture: str, stage_dir: Path, output_path: Path +) -> None: + validate_stage(metadata_path, architecture, stage_dir) + output_path.parent.mkdir(parents=True, exist_ok=True) + with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9) as archive: + for path in sorted(stage_dir.rglob("*")): + if path.is_file(): + archive.write(path, path.relative_to(stage_dir).as_posix()) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Package Herdr with Microsoft's pinned ConPTY runtime") + parser.add_argument("--metadata", type=Path, default=DEFAULT_METADATA) + subparsers = parser.add_subparsers(dest="command", required=True) + + stage = subparsers.add_parser("stage") + stage.add_argument("--architecture", choices=("x86_64",), default="x86_64") + stage.add_argument("--package", type=Path, required=True) + stage.add_argument("--herdr-exe", type=Path, required=True) + stage.add_argument("--output-dir", type=Path, required=True) + + archive = subparsers.add_parser("archive") + archive.add_argument("--architecture", choices=("x86_64",), default="x86_64") + archive.add_argument("--stage-dir", type=Path, required=True) + archive.add_argument("--output", type=Path, required=True) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + if args.command == "stage": + stage_bundle( + args.metadata, + args.architecture, + args.package, + args.herdr_exe, + args.output_dir, + ) + else: + archive_bundle(args.metadata, args.architecture, args.stage_dir, args.output) + + +if __name__ == "__main__": + main() -- cgit v1.2.3