from __future__ import annotations from pathlib import Path import urllib.request import hashlib import xml.etree.ElementTree as ET MAVEN_BASE = "https://repo1.maven.org/maven2/org/mustangproject/Mustang-CLI" VERSION_FILE = "VERSION.txt" class SetupError(RuntimeError): pass def _download(url: str, dest: Path, logger) -> None: logger.info("Downloading: %s", url) dest.parent.mkdir(parents=True, exist_ok=True) try: with urllib.request.urlopen(url, timeout=60) as r: data = r.read() dest.write_bytes(data) except Exception as e: raise SetupError(f"Failed to download {url}: {e}") from e def _sha1_of_file(path: Path) -> str: h = hashlib.sha1() with path.open("rb") as f: for chunk in iter(lambda: f.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() def get_latest_mustang_version(logger) -> str: url = f"{MAVEN_BASE}/maven-metadata.xml" logger.info("Checking latest Mustang version from Maven metadata: %s", url) try: with urllib.request.urlopen(url, timeout=30) as r: xml_data = r.read() root = ET.fromstring(xml_data) latest = root.findtext("./versioning/latest") or root.findtext("./versioning/release") if not latest: # fallback: last version in versions list versions = [v.text for v in root.findall("./versioning/versions/version") if v.text] if versions: latest = versions[-1] if not latest: raise SetupError("Could not determine latest version from maven-metadata.xml") return latest.strip() except Exception as e: raise SetupError(f"Failed to read Maven metadata: {e}") from e def read_installed_version(mustang_dir: Path) -> str | None: vf = mustang_dir / VERSION_FILE if vf.exists(): return vf.read_text(encoding="utf-8").strip() or None return None def write_installed_version(mustang_dir: Path, version: str) -> None: mustang_dir.mkdir(parents=True, exist_ok=True) (mustang_dir / VERSION_FILE).write_text(version.strip(), encoding="utf-8") def install_mustang(version: str, target_path: Path, logger, verify_sha1: bool = True) -> Path: jar_name = f"Mustang-CLI-{version}.jar" jar_url = f"{MAVEN_BASE}/{version}/{jar_name}" sha1_url = f"{jar_url}.sha1" tmp = target_path.with_suffix(".download") _download(jar_url, tmp, logger) if verify_sha1: sha1_tmp = target_path.with_suffix(".sha1") _download(sha1_url, sha1_tmp, logger) expected = sha1_tmp.read_text(encoding="utf-8").strip().split()[0] actual = _sha1_of_file(tmp) if expected.lower() != actual.lower(): tmp.unlink(missing_ok=True) raise SetupError(f"SHA1 mismatch for Mustang jar. expected={expected} actual={actual}") sha1_tmp.unlink(missing_ok=True) logger.info("SHA1 verified for Mustang jar") target_path.parent.mkdir(parents=True, exist_ok=True) tmp.replace(target_path) write_installed_version(target_path.parent, version) logger.info("Installed Mustang CLI: %s (version=%s)", target_path, version) return target_path def check_for_mustang_update(mustang_dir: Path, logger) -> tuple[bool, str | None, str]: installed = read_installed_version(mustang_dir) latest = get_latest_mustang_version(logger) if installed is None: return True, None, latest return (installed.strip() != latest.strip()), installed, latest