#!/usr/bin/env python3 """Install a verified ax release into a user directory. Requires minisign.""" import argparse import hashlib import json import os import platform import re import shutil import subprocess import tarfile import tempfile import urllib.request import uuid from pathlib import Path TRUSTED_PUBLIC_KEY = 'RWRhg+rY855XvRXJad45cJssrasNQelJ5AkiAaBG7wObXQ0dFT92O/mO' MAX_ARCHIVE = 512 * 1024 * 1024 MEMBERS = {"ax", "LICENSE", "NOTICE.md", "THIRD-PARTY-NOTICES.txt"} class HttpsRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): if not newurl.startswith("https://"): raise ValueError("Release redirects require HTTPS") return super().redirect_request(req, fp, code, msg, headers, newurl) def safe_name(value): if not isinstance(value, str) or value != Path(value).name or value in {"", ".", ".."}: raise ValueError("Unsafe release filename") return value def digest(path): with path.open("rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def read_url(url, destination, maximum): if not url.startswith("https://"): raise ValueError("Release downloads require HTTPS") opener = urllib.request.build_opener(HttpsRedirectHandler()) with opener.open(url, timeout=60) as source, destination.open("xb") as output: count = 0 while block := source.read(1024 * 1024): count += len(block) if count > maximum: raise ValueError("Download exceeds its size limit") output.write(block) def verify_metadata(directory, key=None): key = TRUSTED_PUBLIC_KEY if key is None else key subprocess.run(["minisign", "-Vm", str(directory / "SHA256SUMS"), "-x", str(directory / "SHA256SUMS.sig"), "-P", key], check=True, stdout=subprocess.DEVNULL) entries = {} for line in (directory / "SHA256SUMS").read_text().splitlines(): checksum, name = line.split(" ", 1) safe_name(name) if name in entries or not re.fullmatch(r"[a-f0-9]{64}", checksum): raise ValueError("Invalid checksum list") entries[name] = checksum if entries.get("manifest.json") != digest(directory / "manifest.json"): raise ValueError("Release manifest checksum mismatch") return json.loads((directory / "manifest.json").read_text()), entries def unpack(archive, directory): seen = set() total = 0 with tarfile.open(archive, "r:gz") as source: for member in source: if member.name not in MEMBERS or member.name in seen or not member.isfile(): raise ValueError("Unexpected archive member or file type") seen.add(member.name) total += member.size if total > MAX_ARCHIVE: raise ValueError("Extracted release exceeds its size limit") with source.extractfile(member) as data, (directory / member.name).open("xb") as dest: shutil.copyfileobj(data, dest) (directory / member.name).chmod(0o755 if member.name == "ax" else 0o644) if seen != MEMBERS: raise ValueError("Release archive is incomplete") def host_target(): system, machine = platform.system(), platform.machine().lower() if system == "Darwin" and machine in {"arm64", "aarch64"}: return "aarch64-apple-darwin" if system == "Linux" and machine in {"amd64", "x86_64", "arm64", "aarch64"}: if not (os.confstr("CS_GNU_LIBC_VERSION") or "").startswith("glibc "): raise ValueError("This Linux release requires glibc") return ("aarch64" if machine in {"arm64", "aarch64"} else "x86_64") + "-unknown-linux-gnu" raise ValueError("This OS/architecture is not supported by this release") def version_tuple(value): return tuple(int(part) for part in value.split(".")) def check_platform(artifact): if artifact.get("minimum_glibc"): actual = os.confstr("CS_GNU_LIBC_VERSION").split()[1] if version_tuple(actual) < version_tuple(artifact["minimum_glibc"]): raise ValueError("The runner's glibc is older than this release supports") if artifact.get("minimum_macos") and version_tuple(platform.mac_ver()[0]) < version_tuple(artifact["minimum_macos"]): raise ValueError("macOS is older than this release supports") def install(directory, prefix, channel, requested_version=None, base_url=None): if shutil.which("minisign") is None: raise ValueError("Install minisign before using the verified installer") manifest, entries = verify_metadata(directory) version = manifest["version"] if not re.fullmatch(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", version): raise ValueError("Invalid release version") if manifest["product"] != "ax" or manifest["channel"] != channel or requested_version not in (None, version): raise ValueError("Release product, channel or version mismatch") matches = [a for a in manifest["artifacts"] if a["target"] == host_target()] if len(matches) != 1: raise ValueError("Release has no unambiguous artifact for this host") artifact = matches[0] check_platform(artifact) filename = safe_name(artifact["file"]) if entries.get(filename) != artifact["sha256"]: raise ValueError("Artifact is not bound to the signed checksums") archive = directory / filename if base_url: read_url(base_url + "/" + filename, archive, min(artifact["size"], MAX_ARCHIVE)) if archive.stat().st_size != artifact["size"] or digest(archive) != entries[filename]: raise ValueError("Artifact size or checksum mismatch") prefix = prefix.expanduser().absolute() managed = prefix / "lib" / "ax" launcher = prefix / "bin" / "ax" if launcher.exists() or launcher.is_symlink(): if not launcher.is_symlink() or not launcher.resolve().is_relative_to(managed.resolve()): raise ValueError("Existing ax is not managed by this installer; choose another --prefix") if requested_version is None: reported = subprocess.check_output([str(launcher), "--version"], text=True, timeout=10).strip() match = re.fullmatch(r"ax ((?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*))", reported) if match is None: raise ValueError("Cannot determine the installed ax version") if version_tuple(version) < version_tuple(match[1]): raise ValueError("Downgrading ax requires an explicit --version") managed.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(prefix=".stage-", dir=managed) as temp: staged = Path(temp) unpack(archive, staged) if digest(staged / "ax") != artifact["binary_sha256"]: raise ValueError("Extracted executable checksum mismatch") reported = subprocess.check_output([str(staged / "ax"), "--version"], text=True, timeout=10).strip() if reported != "ax " + version: raise ValueError("Candidate executable reports another version") destination = managed / artifact["binary_sha256"] if destination.exists(): if digest(destination / "ax") != artifact["binary_sha256"]: raise ValueError("Existing managed version has unexpected bytes") else: os.rename(staged, destination) staged.mkdir() launcher.parent.mkdir(parents=True, exist_ok=True) temporary_link = launcher.parent / (".ax-" + uuid.uuid4().hex) try: temporary_link.symlink_to(os.path.relpath(destination / "ax", launcher.parent)) os.replace(temporary_link, launcher) finally: temporary_link.unlink(missing_ok=True) return {"version": version, "channel": channel, "installed": str(launcher), "target": artifact["target"]} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--version") parser.add_argument("--channel", choices=["preview", "stable"], default="preview") parser.add_argument("--prefix", type=Path, default=Path.home() / ".local") parser.add_argument("--release-directory", type=Path, help="Verify and install an already downloaded release") args = parser.parse_args() if args.release_directory: result = install(args.release_directory, args.prefix, args.channel, args.version) else: base = "https://downloads.getaster.now/ax/" + args.channel with tempfile.TemporaryDirectory(prefix="ax-download-") as temp: directory = Path(temp) version = args.version if version is None: latest = directory / "latest.json" read_url(base + "/latest.json", latest, 1024 * 1024) version = json.loads(latest.read_text())["version"] if not re.fullmatch(r"(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)", version): raise ValueError("Invalid release version") base += "/" + version for name in ["manifest.json", "SHA256SUMS", "SHA256SUMS.sig"]: read_url(base + "/" + name, directory / name, 1024 * 1024) if json.loads((directory / "manifest.json").read_text())["version"] != version: raise ValueError("Downloaded manifest does not match the selected release") result = install(directory, args.prefix, args.channel, args.version, base) print(json.dumps(result, indent=2)) if __name__ == "__main__": main()