spiral-sdk 0.1.12

"""Download, verify and cache this Rust SDK release's libspiral archive before compiling it.

`spiral-sdk` links the cached archive for its release and target; it never downloads
during a Cargo build. Run this script from a `rust-sdk/X.Y.Z` checkout or an unpacked
SDK crate. The SDK's `native-runtime.json` carries archive checksums or identifies
the immutable public manifest for this release.
New releases download public HTTPS archives without credentials. Older release
records use private GitHub assets, read with `GH_TOKEN`, `GITHUB_TOKEN` or `gh auth token`.
"""

import argparse
import hashlib
import json
import os
import re
import shutil
import subprocess
import tarfile
import tempfile
import urllib.request
from pathlib import Path
from urllib.parse import urlsplit

ARCHIVE_FILES = {"libspiral.a", "bindings.rs", "scandal.h", "link.txt", "LICENSE", "LICENSE-libspiral"}
ROOT = Path(__file__).resolve().parent.parent
RELEASE = ROOT / "native-runtime.json"
if not RELEASE.is_file():
    RELEASE = ROOT / "lang/rust/spiral-sdk/native-runtime.json"


def verify(archive: Path, checksum: str, source: str, target: str, destination: Path) -> dict:
    """Check the archive against its pins and extract its verified files into `destination`."""
    if not re.fullmatch(r"[0-9a-f]{64}", checksum) or not re.fullmatch(r"[0-9a-f]{40}", source):
        raise ValueError("Native archive requires exact SHA-256 and source pins")
    if file_sha256(archive) != checksum:
        raise ValueError("Native archive checksum mismatch")
    with tarfile.open(archive) as tar:
        names = set()
        for member in tar.getmembers():
            path = Path(member.name)
            if path.is_absolute() or ".." in path.parts or not path.parts or path.parts[0] != "native-runtime":
                raise ValueError("Native archive has an unsafe path")
            if member.name in names or (not member.isfile() and not member.isdir()):
                raise ValueError("Native archive contains duplicate paths or non-regular files")
            names.add(member.name)
        tar.extractall(destination, filter="data")
    root = destination / "native-runtime"
    manifest = json.loads((root / "manifest.json").read_text())
    if manifest["schema"] != 1 or manifest["source_sha"] != source or manifest["target"] != target:
        raise ValueError("Native archive target or source mismatch")
    if not re.fullmatch(r"[0-9A-Za-z.+-]+", manifest["version"]):
        raise ValueError("Native archive has an invalid release version")
    if set(manifest["files"]) != ARCHIVE_FILES:
        raise ValueError("Native archive is missing required files")
    expected = {"native-runtime", "native-runtime/manifest.json"}
    expected.update(f"native-runtime/{filename}" for filename in manifest["files"])
    if names - expected:
        raise ValueError("Native archive contains unexpected files")
    for filename, expected in manifest["files"].items():
        if file_sha256(root / filename) != expected:
            raise ValueError(f"Native archive file checksum mismatch: {filename}")
    return manifest


def install(archive: Path, checksum: str, source: str, target: str, cache: Path) -> Path:
    """Verify the archive and replace the cached files for its release and target."""
    with tempfile.TemporaryDirectory() as staging:
        manifest = verify(archive, checksum, source, target, Path(staging))
        installed = cache / manifest["version"] / target
        if installed.exists():
            shutil.rmtree(installed)
        installed.parent.mkdir(parents=True, exist_ok=True)
        shutil.move(Path(staging) / "native-runtime", installed)
    return installed.resolve()


def release_pin(release: Path, target: str) -> tuple[dict, dict]:
    """This checkout's release record and its archive pin for `target`."""
    if not release.is_file():
        raise ValueError("This checkout is not a Rust SDK release; use a rust-sdk/X.Y.Z tag")
    record = json.loads(release.read_text())
    if "manifest_url" in record:
        version = record["version"]
        url = f"https://downloads.spiraldb.com/libspiral/{version}/native-runtime.json"
        if not re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+", version) or record["manifest_url"] != url:
            raise ValueError("Native release requires its version's public manifest URL")
        with urllib.request.urlopen(url, timeout=120) as response:
            published = json.load(response)
        if any(published.get(key) != record[key] for key in ("schema", "version", "source_sha", "tag", "repository")):
            raise ValueError("Public native manifest differs from this SDK release")
        record = published
    if record["schema"] != 1 or target not in record["archives"]:
        raise ValueError(f"Rust SDK {record['version']} has no native archive for {target}")
    return record, record["archives"][target]


def download(repository: str, tag: str, asset: str, directory: Path) -> Path:
    """Download a release asset with a token that can read the repository's contents."""
    if not re.fullmatch(r"[\w.-]+/[\w.-]+", repository) or not re.fullmatch(r"[\w.+-]+\.tar\.gz", asset):
        raise ValueError("Native archive requires an owner/name repository and an archive name")
    token = os.environ.get("GH_TOKEN") or os.environ.get("GITHUB_TOKEN")
    if not token:
        token = subprocess.check_output(["gh", "auth", "token"], text=True).strip()

    def request(url: str, accept: str) -> urllib.request.Request:
        request = urllib.request.Request(url, headers={"Accept": accept})
        # The asset API redirects to blob storage, which must not receive the token.
        request.add_unredirected_header("Authorization", f"Bearer {token}")
        return request

    release = request(f"https://api.github.com/repos/{repository}/releases/tags/{tag}", "application/vnd.github+json")
    with urllib.request.urlopen(release) as response:
        assets = {entry["name"]: entry["url"] for entry in json.load(response)["assets"]}
    if asset not in assets:
        raise ValueError(f"Release {tag} has no asset {asset}")
    archive = directory / asset
    with urllib.request.urlopen(request(assets[asset], "application/octet-stream")) as response:
        with archive.open("wb") as output:
            shutil.copyfileobj(response, output)
    return archive


def download_public(url: str, asset: str, directory: Path) -> Path:
    """Download a public HTTPS release asset without consulting GitHub credentials."""
    parsed = urlsplit(url)
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.query
        or parsed.fragment
        or not re.fullmatch(r"[\w.+-]+\.(?:tar\.gz|json|txt)", asset)
        or parsed.path.rsplit("/", 1)[-1] != asset
    ):
        raise ValueError("Native archive requires a public HTTPS URL matching its archive name")
    archive = directory / asset
    with urllib.request.urlopen(url, timeout=120) as response, archive.open("wb") as output:
        shutil.copyfileobj(response, output)
    return archive


def download_release(record: dict, pin: dict, directory: Path) -> Path:
    """Use the release's public URL, or its legacy private GitHub asset."""
    if "url" in pin:
        return download_public(pin["url"], pin["asset"], directory)
    return download(record["repository"], record["tag"], pin["asset"], directory)


def cache_directory() -> Path:
    """The cache `spiral-sdk` reads, keyed by release version and target."""
    if directory := os.environ.get("SPIRAL_NATIVE_CACHE_DIR"):
        return Path(directory)
    base = os.environ.get("XDG_CACHE_HOME") or Path.home() / ".cache"
    return Path(base) / "spiral" / "native"


def host_target() -> str:
    output = subprocess.check_output(["rustc", "-vV"], text=True)
    return next(line.split(": ", 1)[1] for line in output.splitlines() if line.startswith("host: "))


def file_sha256(path: Path) -> str:
    with path.open("rb") as source:
        return hashlib.file_digest(source, "sha256").hexdigest()


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--target", help="Rust target triple; defaults to the host")
    parser.add_argument("--cache-dir", type=Path, default=None)
    args = parser.parse_args()
    target = args.target or host_target()
    record, pin = release_pin(RELEASE, target)
    cache = args.cache_dir or cache_directory()
    with tempfile.TemporaryDirectory() as downloads:
        archive = download_release(record, pin, Path(downloads))
        installed = install(archive, pin["sha256"], record["source_sha"], target, cache)
    print(installed)
    # In GitHub Actions, expose the prepared directory to later steps.
    if output := os.environ.get("GITHUB_OUTPUT"):
        with Path(output).open("a") as file:
            file.write(f"path={installed}\n")
    if environment := os.environ.get("GITHUB_ENV"):
        with Path(environment).open("a") as file:
            file.write(f"SPIRAL_NATIVE_DIR={installed}\n")


if __name__ == "__main__":
    main()