rustpad/scripts/update_browser_libs.py

#!/usr/bin/env python3
"""Download and refresh RustPad's third-party browser libraries.

The script uses only the Python standard library. By default it resolves the
current stable package versions from npm. Reproducible builds can use a
committed lock file generated with ``--update-lock`` and consumed with
``--locked``.
"""

from __future__ import annotations

import argparse
import base64
import binascii
import hashlib
import json
import os
import shutil
import sys
import tarfile
import tempfile
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Callable, Iterable

USER_AGENT = "RustPad browser-library updater/0.1"
DEFAULT_TIMEOUT = 45
LOCK_SCHEMA = 1
DEFAULT_LOCK_FILE = Path("scripts/browser-libs.lock.json")


class UpdateError(RuntimeError):
    pass


@dataclass(frozen=True)
class Library:
    key: str
    package: str
    destination: str
    entrypoint: str
    repository_fragment: str
    installer: Callable[[tarfile.TarFile, Path], None]


def registry_url(package: str, version: str = "latest") -> str:
    encoded_package = urllib.parse.quote(package, safe="")
    encoded_version = urllib.parse.quote(version, safe="")
    return f"https://registry.npmjs.org/{encoded_package}/{encoded_version}"


def request_bytes(url: str, timeout: int) -> bytes:
    request = urllib.request.Request(
        url,
        headers={
            "Accept": "application/json, application/octet-stream;q=0.9, */*;q=0.8",
            "User-Agent": USER_AGENT,
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout) as response:
            return response.read()
    except (urllib.error.URLError, TimeoutError, OSError) as error:
        raise UpdateError(f"Cannot download {url}: {error}") from error


def package_metadata(package: str, timeout: int, version: str = "latest") -> dict:
    raw = request_bytes(registry_url(package, version), timeout)
    try:
        metadata = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UpdateError(f"Invalid npm metadata for {package}@{version}") from error
    if not isinstance(metadata, dict):
        raise UpdateError(f"Unexpected npm metadata for {package}@{version}")
    return metadata


def repository_url(metadata: dict) -> str:
    repository = metadata.get("repository", "")
    if isinstance(repository, dict):
        repository = repository.get("url", "")
    return str(repository or "")


def package_dist(metadata: dict) -> dict:
    dist = metadata.get("dist")
    if not isinstance(dist, dict):
        return {}
    return dist


def validate_tarball_url(url: str, package: str) -> None:
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme != "https" or not parsed.netloc:
        raise UpdateError(f"Invalid HTTPS tarball URL for {package}: {url or 'not provided'}")


def validate_metadata(
    library: Library,
    metadata: dict,
    *,
    require_integrity: bool = True,
) -> tuple[str, dict, str]:
    version = str(metadata.get("version") or "").strip()
    dist = package_dist(metadata)
    tarball = str(dist.get("tarball") or "").strip()
    integrity = str(dist.get("integrity") or "").strip()
    shasum = str(dist.get("shasum") or "").strip()
    repository = repository_url(metadata).strip()

    if not version or not tarball:
        raise UpdateError(
            f"npm metadata for {library.package} is missing version or tarball data"
        )
    validate_tarball_url(tarball, library.package)
    if require_integrity and not integrity and not shasum:
        raise UpdateError(f"npm metadata for {library.package}@{version} has no integrity hash")
    if library.repository_fragment.lower() not in repository.lower():
        raise UpdateError(
            f"Unexpected repository for {library.package}: {repository or 'not provided'}"
        )
    return version, dist, repository


def verify_tarball(data: bytes, dist: dict) -> None:
    integrity = str(dist.get("integrity") or "").strip()
    algorithms = {
        "sha512": hashlib.sha512,
        "sha384": hashlib.sha384,
        "sha256": hashlib.sha256,
    }

    if integrity:
        recognized = False
        for token in integrity.split():
            algorithm, separator, encoded = token.partition("-")
            if not separator or algorithm not in algorithms:
                continue
            recognized = True
            try:
                expected = base64.b64decode(encoded, validate=True)
            except (binascii.Error, ValueError) as error:
                raise UpdateError(f"Invalid {algorithm} integrity value") from error
            actual = algorithms[algorithm](data).digest()
            if actual != expected:
                raise UpdateError(f"Tarball integrity verification failed ({algorithm})")
            return
        if not recognized:
            raise UpdateError("Tarball integrity uses an unsupported hash algorithm")

    shasum = str(dist.get("shasum") or "").strip()
    if shasum:
        if hashlib.sha1(data).hexdigest().lower() != shasum.lower():
            raise UpdateError("Tarball SHA-1 verification failed")
        return

    raise UpdateError("Tarball metadata does not contain a supported integrity hash")


def safe_relative(member_name: str, prefix: tuple[str, ...]) -> PurePosixPath | None:
    path = PurePosixPath(member_name)
    parts = path.parts
    if len(parts) <= len(prefix) or tuple(parts[: len(prefix)]) != prefix:
        return None
    relative = PurePosixPath(*parts[len(prefix) :])
    if relative.is_absolute() or any(part in {"", ".", ".."} for part in relative.parts):
        raise UpdateError(f"Unsafe path in package archive: {member_name}")
    return relative


def write_member(archive: tarfile.TarFile, member: tarfile.TarInfo, destination: Path) -> None:
    if not member.isfile():
        return
    source = archive.extractfile(member)
    if source is None:
        raise UpdateError(f"Cannot read {member.name} from package archive")
    destination.parent.mkdir(parents=True, exist_ok=True)
    with source, destination.open("wb") as output:
        shutil.copyfileobj(source, output)


def copy_license(archive: tarfile.TarFile, destination: Path) -> None:
    names = {
        "package/LICENSE",
        "package/LICENSE.txt",
        "package/LICENSE.md",
        "package/LICENCE",
        "package/LICENCE.txt",
        "package/LICENCE.md",
    }
    member = next((item for item in archive.getmembers() if item.isfile() and item.name in names), None)
    if member is None:
        raise UpdateError("The package archive does not contain a license file")
    write_member(archive, member, destination / "LICENSE.txt")


def install_mermaid(archive: tarfile.TarFile, destination: Path) -> None:
    copied = 0
    for member in archive.getmembers():
        relative = safe_relative(member.name, ("package", "dist"))
        if relative is None or not member.isfile():
            continue
        suffix = relative.suffix.lower()
        is_entrypoint = relative == PurePosixPath("mermaid.esm.min.mjs")
        is_minified_chunk = relative.parts[:2] == ("chunks", "mermaid.esm.min")
        if not (is_entrypoint or is_minified_chunk):
            continue
        if suffix not in {".mjs", ".wasm", ".css"}:
            continue
        write_member(archive, member, destination / Path(*relative.parts))
        copied += 1
    if copied == 0 or not (destination / "mermaid.esm.min.mjs").is_file():
        raise UpdateError("Mermaid browser entrypoint was not found in the npm package")
    copy_license(archive, destination)


def install_highlight(archive: tarfile.TarFile, destination: Path) -> None:
    candidates = {
        "package/highlight.min.js",
        "package/build/highlight.min.js",
    }
    member = next((item for item in archive.getmembers() if item.isfile() and item.name in candidates), None)
    if member is None:
        member = next(
            (
                item
                for item in archive.getmembers()
                if item.isfile() and PurePosixPath(item.name).name == "highlight.min.js"
            ),
            None,
        )
    if member is None:
        raise UpdateError("Highlight.js browser build was not found in the npm package")
    write_member(archive, member, destination / "highlight.min.js")
    copy_license(archive, destination)


LIBRARIES = (
    Library(
        key="mermaid",
        package="mermaid",
        destination="mermaid",
        entrypoint="mermaid.esm.min.mjs",
        repository_fragment="mermaid-js/mermaid",
        installer=install_mermaid,
    ),
    Library(
        key="highlight",
        package="@highlightjs/cdn-assets",
        destination="highlight",
        entrypoint="highlight.min.js",
        repository_fragment="highlightjs/highlight.js",
        installer=install_highlight,
    ),
)

LIBRARIES_BY_KEY = {library.key: library for library in LIBRARIES}


def installed_version(destination: Path) -> str:
    version_file = destination / "VERSION"
    try:
        return version_file.read_text(encoding="utf-8").strip()
    except OSError:
        return ""


def is_complete(library: Library, destination: Path) -> bool:
    return (destination / library.entrypoint).is_file() and (destination / "LICENSE.txt").is_file()


def source_note(library: Library, metadata: dict, version: str) -> str:
    dist = package_dist(metadata)
    repository = repository_url(metadata)
    return (
        f"Package: {library.package}\n"
        f"Version: {version}\n"
        f"Registry: {registry_url(library.package, version)}\n"
        f"Tarball: {dist.get('tarball', '')}\n"
        f"Integrity: {dist.get('integrity') or dist.get('shasum') or ''}\n"
        f"Repository: {repository}\n"
        "Generated by scripts/update_browser_libs.py; do not edit or commit this directory.\n"
    )


def install_library(
    library: Library,
    metadata: dict,
    destination: Path,
    timeout: int,
) -> None:
    version, dist, _ = validate_metadata(library, metadata)
    tarball = request_bytes(str(dist["tarball"]), timeout)
    verify_tarball(tarball, dist)

    destination.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix=f".{library.key}-", dir=destination.parent) as temporary:
        staging = Path(temporary) / library.destination
        staging.mkdir(parents=True)
        archive_path = Path(temporary) / "package.tgz"
        archive_path.write_bytes(tarball)
        try:
            with tarfile.open(archive_path, mode="r:gz") as archive:
                library.installer(archive, staging)
        except (tarfile.TarError, OSError) as error:
            raise UpdateError(f"Cannot unpack {library.package}: {error}") from error

        (staging / "VERSION").write_text(f"{version}\n", encoding="utf-8")
        (staging / "SOURCE.txt").write_text(
            source_note(library, metadata, version), encoding="utf-8"
        )
        if not is_complete(library, staging):
            raise UpdateError(f"Generated {library.key} directory is incomplete")

        old_destination = destination.with_name(f".{destination.name}.old")
        if old_destination.exists():
            shutil.rmtree(old_destination)
        if destination.exists():
            destination.replace(old_destination)
        try:
            shutil.move(str(staging), str(destination))
        except Exception:
            if old_destination.exists() and not destination.exists():
                old_destination.replace(destination)
            raise
        finally:
            if old_destination.exists():
                shutil.rmtree(old_destination)


def selected_libraries(keys: Iterable[str]) -> list[Library]:
    requested = set(keys)
    if not requested:
        return list(LIBRARIES)
    return [library for library in LIBRARIES if library.key in requested]


def resolve_path(root: Path, value: Path) -> Path:
    return value.resolve() if value.is_absolute() else (root / value).resolve()


def load_lock_file(path: Path, *, required: bool) -> dict:
    if not path.is_file():
        if required:
            raise UpdateError(
                f"Lock file does not exist: {path}. Generate it with --update-lock first."
            )
        return {"schema": LOCK_SCHEMA, "libraries": {}}
    try:
        document = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UpdateError(f"Cannot read lock file {path}: {error}") from error
    if not isinstance(document, dict) or document.get("schema") != LOCK_SCHEMA:
        raise UpdateError(f"Unsupported or missing lock-file schema in {path}")
    entries = document.get("libraries")
    if not isinstance(entries, dict):
        raise UpdateError(f"Lock file {path} has no libraries object")
    return document


def metadata_from_lock(library: Library, document: dict) -> dict:
    entries = document["libraries"]
    entry = entries.get(library.key)
    if not isinstance(entry, dict):
        raise UpdateError(f"Lock file has no entry for {library.key}")

    package = str(entry.get("package") or "").strip()
    version = str(entry.get("version") or "").strip()
    tarball = str(entry.get("tarball") or "").strip()
    integrity = str(entry.get("integrity") or "").strip()
    shasum = str(entry.get("shasum") or "").strip()
    repository = str(entry.get("repository") or "").strip()

    if package != library.package:
        raise UpdateError(
            f"Lock entry {library.key} points to {package or 'no package'}, expected {library.package}"
        )
    metadata = {
        "name": package,
        "version": version,
        "repository": repository,
        "dist": {
            "tarball": tarball,
            "integrity": integrity,
            "shasum": shasum,
        },
    }
    validate_metadata(library, metadata)
    return metadata


def lock_entry_from_metadata(library: Library, metadata: dict) -> dict:
    version, dist, repository = validate_metadata(library, metadata)
    return {
        "package": library.package,
        "version": version,
        "tarball": str(dist.get("tarball") or ""),
        "integrity": str(dist.get("integrity") or ""),
        "shasum": str(dist.get("shasum") or ""),
        "repository": repository,
    }


def write_lock_file(path: Path, document: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = json.dumps(document, indent=2, sort_keys=True, ensure_ascii=True) + "\n"
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
    )
    temporary_path = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as output:
            output.write(payload)
            output.flush()
            os.fsync(output.fileno())
        os.replace(temporary_path, path)
    except Exception:
        temporary_path.unlink(missing_ok=True)
        raise


def parse_version_overrides(values: Iterable[str]) -> dict[str, str]:
    overrides: dict[str, str] = {}
    for value in values:
        key, separator, version = value.partition("=")
        key = key.strip()
        version = version.strip()
        if not separator or key not in LIBRARIES_BY_KEY or not version:
            choices = ", ".join(sorted(LIBRARIES_BY_KEY))
            raise UpdateError(
                f"Invalid --version value {value!r}; expected LIBRARY=VERSION ({choices})"
            )
        if key in overrides:
            raise UpdateError(f"Duplicate --version value for {key}")
        overrides[key] = version
    return overrides


def update_one(
    library: Library,
    metadata: dict,
    libs_root: Path,
    *,
    timeout: int,
    force: bool,
    check: bool,
) -> bool:
    target, _, _ = validate_metadata(library, metadata)
    destination = libs_root / library.destination
    current = installed_version(destination)
    complete = is_complete(library, destination)

    if complete and current == target and not force:
        print(f"{library.key}: up to date ({target})")
        return True

    state = "missing" if not complete else f"{current or 'unknown'} -> {target}"
    if check:
        print(f"{library.key}: update required ({state})")
        return False

    print(f"{library.key}: downloading {target} ({state})")
    install_library(library, metadata, destination, timeout)
    print(f"{library.key}: installed {target}")
    return True


def check_updates(
    libraries: list[Library],
    libs_root: Path,
    lock_path: Path,
    timeout: int,
) -> bool:
    document = load_lock_file(lock_path, required=False)
    entries = document["libraries"]
    success = True

    for library in libraries:
        baseline = ""
        source = "installed copy"
        if library.key in entries:
            baseline = str(metadata_from_lock(library, document).get("version") or "")
            source = "lock file"
        if not baseline:
            baseline = installed_version(libs_root / library.destination)

        latest_metadata = package_metadata(library.package, timeout)
        latest, _, _ = validate_metadata(library, latest_metadata)
        if not baseline:
            print(f"{library.key}: no locked or installed version; latest is {latest}")
            success = False
        elif baseline == latest:
            print(f"{library.key}: {source} is current ({latest})")
        else:
            print(f"{library.key}: update available ({baseline} -> {latest}, based on {source})")
            success = False
    return success


def parse_args() -> argparse.Namespace:
    script_root = Path(__file__).resolve().parent.parent
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--root", type=Path, default=script_root, help="RustPad repository root")
    parser.add_argument(
        "--library",
        action="append",
        choices=[item.key for item in LIBRARIES],
        default=[],
        help="limit the operation to one library; may be repeated",
    )
    parser.add_argument("--force", action="store_true", help="download again even when the version is current")
    parser.add_argument("--check", action="store_true", help="only verify that installed assets match the target versions")
    parser.add_argument("--strict", action="store_true", help="fail when the registry cannot be reached")
    parser.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT, help="network timeout in seconds")
    parser.add_argument(
        "--lock-file",
        type=Path,
        default=DEFAULT_LOCK_FILE,
        help=f"lock-file path relative to --root (default: {DEFAULT_LOCK_FILE})",
    )
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument(
        "--locked",
        action="store_true",
        help="install exact versions and tarballs from the lock file without querying npm metadata",
    )
    modes.add_argument(
        "--update-lock",
        action="store_true",
        help="resolve target versions, install them, and atomically update the lock file",
    )
    modes.add_argument(
        "--check-updates",
        action="store_true",
        help="compare locked or installed versions with the current npm latest versions",
    )
    parser.add_argument(
        "--version",
        action="append",
        default=[],
        metavar="LIBRARY=VERSION",
        help="resolve an exact npm version instead of latest; may be repeated",
    )
    return parser.parse_args()


def validate_arguments(args: argparse.Namespace, overrides: dict[str, str]) -> None:
    if args.timeout < 1:
        raise UpdateError("--timeout must be at least 1 second")
    if args.locked and overrides:
        raise UpdateError("--version cannot be combined with --locked")
    if args.check_updates and (args.check or args.force or overrides):
        raise UpdateError("--check-updates cannot be combined with --check, --force, or --version")
    if args.update_lock and args.check:
        raise UpdateError("--update-lock cannot be combined with --check")
    selected = set(args.library)
    if selected:
        outside_selection = sorted(set(overrides) - selected)
        if outside_selection:
            raise UpdateError(
                "--version was provided for an unselected library: "
                + ", ".join(outside_selection)
            )


def main() -> int:
    args = parse_args()
    try:
        overrides = parse_version_overrides(args.version)
        validate_arguments(args, overrides)

        root = args.root.resolve()
        libs_root = root / "static" / "libs"
        lock_path = resolve_path(root, args.lock_file)
        libraries = selected_libraries(args.library)

        if args.check_updates:
            return 0 if check_updates(libraries, libs_root, lock_path, args.timeout) else 1

        if args.locked or args.update_lock:
            lock_document = load_lock_file(lock_path, required=args.locked)
        else:
            lock_document = {"schema": LOCK_SCHEMA, "libraries": {}}
        targets: dict[str, dict] = {}
        skipped: set[str] = set()
        effective_strict = args.strict or args.check or args.update_lock

        for library in libraries:
            if args.locked:
                targets[library.key] = metadata_from_lock(library, lock_document)
                continue

            requested_version = overrides.get(library.key, "latest")
            try:
                targets[library.key] = package_metadata(
                    library.package,
                    args.timeout,
                    requested_version,
                )
                validate_metadata(library, targets[library.key])
            except UpdateError as error:
                destination = libs_root / library.destination
                complete = is_complete(library, destination)
                current = installed_version(destination)
                if complete and not effective_strict:
                    print(
                        f"warning: {error}; keeping {library.key} {current or 'local copy'}",
                        file=sys.stderr,
                    )
                    skipped.add(library.key)
                    continue
                raise

        success = True
        for library in libraries:
            if library.key in skipped:
                continue
            success = update_one(
                library,
                targets[library.key],
                libs_root,
                timeout=args.timeout,
                force=args.force,
                check=args.check,
            ) and success

        if args.update_lock and success:
            entries = dict(lock_document["libraries"])
            for library in libraries:
                entries[library.key] = lock_entry_from_metadata(library, targets[library.key])
            updated_document = {
                "schema": LOCK_SCHEMA,
                "libraries": entries,
            }
            write_lock_file(lock_path, updated_document)
            print(f"lock: updated {lock_path}")

        return 0 if success else 1
    except UpdateError as error:
        print(f"error: {error}", file=sys.stderr)
        return 1
    except OSError as error:
        print(f"error: {error}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())