Build verified Linux release packages

tools/install_godot_templates.py

@@ -0,0 +1,221 @@
+#!/usr/bin/env python3
+"""Install selected, pinned Godot export templates with HTTP range requests.
+
+Godot 4.7 stores every platform template in one archive. This reader mirrors the
+editor's selective downloader: ZIP metadata and only the requested compressed
+members cross the network. Each installed member is then checked against the
+project manifest's size and SHA-256.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import io
+import json
+import os
+import time
+import urllib.error
+import urllib.request
+import zipfile
+from collections.abc import Callable
+from pathlib import Path
+
+DEFAULT_WINDOW = 8 * 1024 * 1024
+HTTP_HEADERS = {
+    "User-Agent": "OpenMaidEngine-build/1.0 (+https://git.orfl.xyz/conco/OpenMaidEngine)",
+    "Accept": "application/octet-stream,*/*;q=0.8",
+}
+
+
+class RangeReader(io.RawIOBase):
+    """Seekable read-only view backed by a byte-range callback."""
+
+    def __init__(
+        self,
+        length: int,
+        fetch: Callable[[int, int], bytes],
+        window_size: int = DEFAULT_WINDOW,
+    ) -> None:
+        if length <= 0:
+            raise ValueError("range source length must be positive")
+        self._length = length
+        self._fetch = fetch
+        self._window_size = window_size
+        self._position = 0
+        self._cache_start = 0
+        self._cache = b""
+
+    def readable(self) -> bool:
+        return True
+
+    def seekable(self) -> bool:
+        return True
+
+    def tell(self) -> int:
+        return self._position
+
+    def seek(self, offset: int, whence: int = io.SEEK_SET) -> int:
+        if whence == io.SEEK_SET:
+            position = offset
+        elif whence == io.SEEK_CUR:
+            position = self._position + offset
+        elif whence == io.SEEK_END:
+            position = self._length + offset
+        else:
+            raise ValueError(f"unsupported seek mode: {whence}")
+        if position < 0:
+            raise ValueError("negative seek position")
+        self._position = position
+        return position
+
+    def read(self, size: int = -1) -> bytes:
+        if self._position >= self._length:
+            return b""
+        if size is None or size < 0:
+            size = self._length - self._position
+        size = min(size, self._length - self._position)
+        if size == 0:
+            return b""
+
+        cache_end = self._cache_start + len(self._cache)
+        requested_end = self._position + size
+        if not (
+            self._cache_start <= self._position
+            and requested_end <= cache_end
+        ):
+            fetch_size = max(size, self._window_size)
+            fetch_end = min(self._length, self._position + fetch_size)
+            self._cache_start = self._position
+            self._cache = self._fetch(self._position, fetch_end - 1)
+            expected = fetch_end - self._position
+            if len(self._cache) != expected:
+                raise OSError(
+                    f"short range response: expected {expected} bytes, "
+                    f"received {len(self._cache)}"
+                )
+            cache_end = fetch_end
+
+        offset = self._position - self._cache_start
+        data = self._cache[offset:offset + size]
+        self._position += len(data)
+        return data
+
+
+def _open_with_retry(request: urllib.request.Request, attempts: int = 3):
+    last_error: Exception | None = None
+    for attempt in range(attempts):
+        try:
+            return urllib.request.urlopen(request, timeout=120)
+        except (OSError, urllib.error.URLError) as error:
+            last_error = error
+            if attempt + 1 < attempts:
+                time.sleep(2 ** attempt)
+    assert last_error is not None
+    raise last_error
+
+
+def open_http_range_reader(url: str, expected_length: int) -> RangeReader:
+    head = urllib.request.Request(url, headers=HTTP_HEADERS, method="HEAD")
+    with _open_with_retry(head) as response:
+        final_url = response.geturl()
+        actual_length = int(response.headers.get("Content-Length", "0"))
+    if actual_length != expected_length:
+        raise ValueError(
+            f"template archive size mismatch: expected {expected_length}, "
+            f"server reported {actual_length}"
+        )
+
+    def fetch(start: int, end: int) -> bytes:
+        request_headers = dict(HTTP_HEADERS)
+        request_headers.update({
+            "Range": f"bytes={start}-{end}",
+            "Accept-Encoding": "identity",
+        })
+        request = urllib.request.Request(final_url, headers=request_headers)
+        with _open_with_retry(request) as response:
+            if response.status != 206:
+                raise OSError(
+                    f"template server ignored byte range {start}-{end}: "
+                    f"HTTP {response.status}"
+                )
+            content_range = response.headers.get("Content-Range", "")
+            expected_range = f"bytes {start}-{end}/{expected_length}"
+            if content_range != expected_range:
+                raise OSError(
+                    f"unexpected Content-Range: expected {expected_range!r}, "
+                    f"received {content_range!r}"
+                )
+            return response.read()
+
+    return RangeReader(expected_length, fetch)
+
+
+def file_matches(path: Path, expected_size: int, expected_hash: str) -> bool:
+    if not path.is_file() or path.stat().st_size != expected_size:
+        return False
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for block in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest() == expected_hash.lower()
+
+
+def install_members(manifest_path: Path, destination: Path) -> list[Path]:
+    with manifest_path.open(encoding="utf-8") as stream:
+        manifest = json.load(stream)
+    templates = manifest["templates"]
+    members = templates["members"]
+    destination.mkdir(parents=True, exist_ok=True)
+
+    installed = [destination / member["install_name"] for member in members]
+    if all(
+        file_matches(path, member["size"], member["sha256"])
+        for path, member in zip(installed, members, strict=True)
+    ):
+        return installed
+
+    reader = open_http_range_reader(templates["url"], templates["size"])
+    with zipfile.ZipFile(reader) as archive:
+        for path, member in zip(installed, members, strict=True):
+            if file_matches(path, member["size"], member["sha256"]):
+                continue
+            archive_path = member["archive_path"]
+            info = archive.getinfo(archive_path)
+            if info.file_size != member["size"]:
+                raise ValueError(
+                    f"template member size mismatch for {archive_path}: "
+                    f"expected {member['size']}, archive reports {info.file_size}"
+                )
+            temporary = path.with_suffix(path.suffix + ".part")
+            digest = hashlib.sha256()
+            with archive.open(info) as source, temporary.open("wb") as target:
+                for block in iter(lambda: source.read(1024 * 1024), b""):
+                    target.write(block)
+                    digest.update(block)
+            if temporary.stat().st_size != member["size"]:
+                temporary.unlink(missing_ok=True)
+                raise ValueError(f"short extracted template member: {archive_path}")
+            actual_hash = digest.hexdigest()
+            if actual_hash != member["sha256"].lower():
+                temporary.unlink(missing_ok=True)
+                raise ValueError(
+                    f"template member SHA-256 mismatch for {archive_path}: "
+                    f"expected {member['sha256']}, got {actual_hash}"
+                )
+            os.chmod(temporary, int(member["mode"], 8))
+            os.replace(temporary, path)
+    return installed
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--manifest", type=Path, required=True)
+    parser.add_argument("--destination", type=Path, required=True)
+    args = parser.parse_args()
+    for installed in install_members(args.manifest.resolve(), args.destination.resolve()):
+        print(installed)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())