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. It resolves the current
stable package version from the npm registry, verifies the downloaded tarball,
and atomically replaces the generated directory under static/libs.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
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/1.0"
DEFAULT_TIMEOUT = 45


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) -> str:
    encoded = urllib.parse.quote(package, safe="")
    return f"https://registry.npmjs.org/{encoded}/latest"


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) -> dict:
    raw = request_bytes(registry_url(package), timeout)
    try:
        metadata = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError) as error:
        raise UpdateError(f"Invalid npm metadata for {package}") from error
    if not isinstance(metadata, dict):
        raise UpdateError(f"Unexpected npm metadata for {package}")
    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 verify_tarball(data: bytes, dist: dict) -> None:
    integrity = str(dist.get("integrity") or "")
    if integrity:
        algorithms = {
            "sha512": hashlib.sha512,
            "sha384": hashlib.sha384,
            "sha256": hashlib.sha256,
        }
        for token in integrity.split():
            algorithm, separator, encoded = token.partition("-")
            if not separator or algorithm not in algorithms:
                continue
            expected = base64.b64decode(encoded)
            actual = algorithms[algorithm](data).digest()
            if actual != expected:
                raise UpdateError(f"Tarball integrity verification failed ({algorithm})")
            return
    shasum = str(dist.get("shasum") or "")
    if shasum and hashlib.sha1(data).hexdigest().lower() != shasum.lower():
        raise UpdateError("Tarball SHA-1 verification failed")


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,
    ),
)


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:
    repository = repository_url(metadata)
    return (
        f"Package: {library.package}\n"
        f"Version: {version}\n"
        f"Registry: {registry_url(library.package)}\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 = str(metadata.get("version") or "").strip()
    dist = metadata.get("dist")
    if not version or not isinstance(dist, dict) or not dist.get("tarball"):
        raise UpdateError(f"npm metadata for {library.package} is missing version or tarball data")

    repository = repository_url(metadata).lower()
    if library.repository_fragment.lower() not in repository:
        raise UpdateError(
            f"Unexpected repository for {library.package}: {repository or 'not provided'}"
        )

    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 update_one(
    library: Library,
    libs_root: Path,
    *,
    timeout: int,
    force: bool,
    check: bool,
    strict: bool,
) -> bool:
    destination = libs_root / library.destination
    current = installed_version(destination)
    complete = is_complete(library, destination)

    try:
        metadata = package_metadata(library.package, timeout)
    except UpdateError as error:
        if complete and not strict:
            print(f"warning: {error}; keeping {library.key} {current or 'local copy'}", file=sys.stderr)
            return True
        raise

    latest = str(metadata.get("version") or "").strip()
    if not latest:
        raise UpdateError(f"npm did not return a version for {library.package}")

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

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

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


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=[])
    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 report whether an update is required")
    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")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    root = args.root.resolve()
    libs_root = root / "static" / "libs"
    if args.timeout < 1:
        print("error: --timeout must be at least 1 second", file=sys.stderr)
        return 2

    success = True
    try:
        for library in selected_libraries(args.library):
            success = update_one(
                library,
                libs_root,
                timeout=args.timeout,
                force=args.force,
                check=args.check,
                strict=args.strict or args.check,
            ) and success
    except UpdateError as error:
        print(f"error: {error}", file=sys.stderr)
        return 1
    return 0 if success else 1


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