Add client zip download helper

update_build_metadata.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+import hashlib
+import json
+import os
+import re
+import shutil
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime
+from pathlib import Path
+
+try:
+    import tomllib  # Python 3.11+
+except ModuleNotFoundError:  # pragma: no cover - fallback for older runtimes
+    try:
+        import tomli as tomllib  # type: ignore
+    except ModuleNotFoundError:  # pragma: no cover
+        tomllib = None  # type: ignore
+
+
+def fetch_manifest(url: str) -> dict | None:
+    try:
+        with urllib.request.urlopen(url, timeout=15) as resp:
+            return json.load(resp)
+    except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
+        print(f"[WARN] Failed to download manifest from {url}: {exc}", file=sys.stderr)
+        return None
+
+
+def select_latest_build(builds: dict) -> dict | None:
+    if not builds:
+        return None
+    # Builds are keyed by commit hash; pick the newest by the ISO8601 timestamp.
+    def build_time(item):
+        _, data = item
+        try:
+            return datetime.fromisoformat(data.get("time", "").rstrip("Z"))
+        except ValueError:
+            return datetime.min
+
+    latest_key, latest_data = max(builds.items(), key=build_time)
+    client = latest_data.get("client")
+    if not client:
+        print(f"[WARN] Manifest entry {latest_key} is missing client info", file=sys.stderr)
+        return None
+    client.setdefault("version", latest_key)
+    return client
+
+
+def update_config(config_path: str, url: str, sha256: str) -> bool:
+    try:
+        config_text = Path(config_path).read_text(encoding="utf-8")
+    except OSError as exc:
+        print(f"[WARN] Cannot read config {config_path}: {exc}", file=sys.stderr)
+        return False
+
+    download_pattern = re.compile(r"(?m)^(#\s*)?download_url\s*=.*$")
+    build_pattern = re.compile(r"(?m)^(#\s*)?build\s*=.*$")
+
+    new_text = download_pattern.sub(f'download_url = "{url}"', config_text, count=1)
+    new_text = build_pattern.sub(f'build = "{sha256.upper()}"', new_text, count=1)
+
+    if new_text == config_text:
+        print("[WARN] No download_url/build entries were updated in server_config.toml", file=sys.stderr)
+        return False
+
+    try:
+        Path(config_path).write_text(new_text, encoding="utf-8")
+    except OSError as exc:
+        print(f"[WARN] Cannot write config {config_path}: {exc}", file=sys.stderr)
+        return False
+
+    print(f"[INFO] Updated build.download_url to {url}")
+    return True
+
+
+def parse_build_section(config_path: str) -> tuple[str | None, str | None] | None:
+    if tomllib is None:
+        print("[WARN] tomllib/tomli is not available; cannot parse server_config.toml", file=sys.stderr)
+        return None
+
+    try:
+        contents = Path(config_path).read_text(encoding="utf-8")
+    except OSError as exc:
+        print(f"[WARN] Cannot read config {config_path}: {exc}", file=sys.stderr)
+        return None
+
+    try:
+        data = tomllib.loads(contents)
+    except (tomllib.TOMLDecodeError, AttributeError) as exc:  # type: ignore[attr-defined]
+        print(f"[WARN] Failed to parse TOML {config_path}: {exc}", file=sys.stderr)
+        return None
+
+    build_section = data.get("build")
+    if not isinstance(build_section, dict):
+        print("[WARN] No [build] section found in server_config.toml", file=sys.stderr)
+        return None
+
+    download_url = build_section.get("download_url") or build_section.get("download-url")
+    build_hash = build_section.get("build")
+
+    download_url = download_url.strip() if isinstance(download_url, str) else None
+    build_hash = build_hash.strip() if isinstance(build_hash, str) else None
+
+    return download_url, build_hash
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as file_obj:
+        for chunk in iter(lambda: file_obj.read(65536), b""):
+            digest.update(chunk)
+    return digest.hexdigest().upper()
+
+
+def download_to_path(url: str, dest: Path) -> bool:
+    tmp_path = dest.with_suffix(dest.suffix + ".tmp")
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        with urllib.request.urlopen(url, timeout=60) as resp, tmp_path.open("wb") as out_file:
+            shutil.copyfileobj(resp, out_file)
+        tmp_path.replace(dest)
+        return True
+    except (urllib.error.URLError, TimeoutError, OSError) as exc:
+        print(f"[WARN] Failed to download {url}: {exc}", file=sys.stderr)
+        try:
+            tmp_path.unlink(missing_ok=True)  # type: ignore[call-arg]
+        except TypeError:
+            if tmp_path.exists():
+                tmp_path.unlink()
+        return False
+
+
+def ensure_client_zip(config_path: str) -> bool:
+    parsed = parse_build_section(config_path)
+    if not parsed:
+        return False
+
+    download_url, expected_hash = parsed
+    if not download_url:
+        print("[WARN] build.download_url is not set; Content.Client.zip cannot be downloaded", file=sys.stderr)
+        return False
+
+    client_zip = Path(config_path).with_name("Content.Client.zip")
+    expected_hash_upper = expected_hash.upper() if expected_hash else None
+
+    if client_zip.exists() and expected_hash_upper:
+        try:
+            current_hash = sha256_file(client_zip)
+        except OSError as exc:
+            print(f"[WARN] Cannot read {client_zip}: {exc}", file=sys.stderr)
+        else:
+            if current_hash == expected_hash_upper:
+                print("[INFO] Content.Client.zip already matches configured build hash")
+                return True
+            print("[INFO] Content.Client.zip hash mismatch; re-downloading")
+
+    if not download_to_path(download_url, client_zip):
+        return False
+
+    if expected_hash_upper:
+        try:
+            downloaded_hash = sha256_file(client_zip)
+        except OSError as exc:
+            print(f"[WARN] Cannot read downloaded {client_zip}: {exc}", file=sys.stderr)
+            return False
+        if downloaded_hash != expected_hash_upper:
+            print(
+                f"[WARN] Downloaded Content.Client.zip hash {downloaded_hash} does not match expected {expected_hash_upper}",
+                file=sys.stderr,
+            )
+            client_zip.unlink(missing_ok=True)  # type: ignore[call-arg]
+            return False
+
+    print(f"[INFO] Downloaded Content.Client.zip to {client_zip}")
+    return True
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2:
+        print("Usage: update_build_metadata.py <path-to-server_config.toml>", file=sys.stderr)
+        sys.exit(1)
+
+    manifest_url = os.environ.get(
+        "ROBUST_CDN_MANIFEST", "https://cdn.wylab.me/fork/wylab/manifest"
+    )
+
+    manifest = fetch_manifest(manifest_url)
+    if not manifest:
+        sys.exit(0)
+
+    client_info = select_latest_build(manifest.get("builds", {}))
+    if not client_info:
+        sys.exit(0)
+
+    client_url = client_info.get("url")
+    client_hash = client_info.get("sha256")
+    if not (client_url and client_hash):
+        print("[WARN] Client entry missing url/sha256", file=sys.stderr)
+        sys.exit(0)
+
+    config_path = sys.argv[1]
+    if update_config(config_path, client_url, client_hash):
+        ensure_client_zip(config_path)
+    else:
+        print("[WARN] Skipping client download because config update failed", file=sys.stderr)