diff --git a/Dockerfile b/Dockerfile index 5e01cbc..18dc716 100644 --- a/Dockerfile +++ b/Dockerfile @@ -3,8 +3,9 @@ WORKDIR /app ARG BROWSER_LIBS_REFRESH=manual COPY scripts/update_browser_libs.py ./scripts/update_browser_libs.py -RUN echo "Browser library refresh: ${BROWSER_LIBS_REFRESH}" \ - && python3 ./scripts/update_browser_libs.py --root /app --strict +COPY scripts/browser-libs.lock.json ./scripts/browser-libs.lock.json +RUN echo "Browser library download: ${BROWSER_LIBS_REFRESH}" \ + && python3 ./scripts/update_browser_libs.py --root /app --locked --strict FROM rust:slim-trixie AS builder WORKDIR /app diff --git a/scripts/browser-libs.lock.json b/scripts/browser-libs.lock.json new file mode 100644 index 0000000..32d22b0 --- /dev/null +++ b/scripts/browser-libs.lock.json @@ -0,0 +1,21 @@ +{ + "libraries": { + "highlight": { + "integrity": "sha512-VEPdHzwelZ12hEX18BHduqxMZGolcUsrbeokHYxOUIm8X2+M7nx5QPtPeQgRxR9XjhdLv4/7DD5BWOlSrJ3k7Q==", + "package": "@highlightjs/cdn-assets", + "repository": "git://github.com/highlightjs/highlight.js.git", + "shasum": "136984ae467865e22080b3a4b65398a086e1ae7b", + "tarball": "https://registry.npmjs.org/@highlightjs/cdn-assets/-/cdn-assets-11.11.1.tgz", + "version": "11.11.1" + }, + "mermaid": { + "integrity": "sha512-TQsq6u22fAn3rek5VOubrhKPo1g5hwC3FXUN9hiyupTckcYiGuuKGkNQrKYwGJkXUxZdojwRG46gsSCFZMDp4g==", + "package": "mermaid", + "repository": "git+https://github.com/mermaid-js/mermaid.git", + "shasum": "57ae2342f6c45b967113b04c9258430bdd057ee8", + "tarball": "https://registry.npmjs.org/mermaid/-/mermaid-11.16.1.tgz", + "version": "11.16.1" + } + }, + "schema": 1 +} diff --git a/scripts/update_browser_libs.py b/scripts/update_browser_libs.py index c73c58d..78f62d4 100755 --- a/scripts/update_browser_libs.py +++ b/scripts/update_browser_libs.py @@ -1,17 +1,20 @@ #!/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. +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 @@ -23,8 +26,10 @@ from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Callable, Iterable -USER_AGENT = "RustPad browser-library updater/1.0" +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): @@ -41,9 +46,10 @@ class Library: 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 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: @@ -61,14 +67,14 @@ def request_bytes(url: str, timeout: int) -> bytes: raise UpdateError(f"Cannot download {url}: {error}") from error -def package_metadata(package: str, timeout: int) -> dict: - raw = request_bytes(registry_url(package), timeout) +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}") from 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}") + raise UpdateError(f"Unexpected npm metadata for {package}@{version}") return metadata @@ -79,26 +85,79 @@ def repository_url(metadata: dict) -> str: 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 "") + integrity = str(dist.get("integrity") or "").strip() + algorithms = { + "sha512": hashlib.sha512, + "sha384": hashlib.sha384, + "sha256": hashlib.sha256, + } + if integrity: - algorithms = { - "sha512": hashlib.sha512, - "sha384": hashlib.sha384, - "sha256": hashlib.sha256, - } + recognized = False for token in integrity.split(): algorithm, separator, encoded = token.partition("-") if not separator or algorithm not in algorithms: continue - expected = base64.b64decode(encoded) + 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 - shasum = str(dist.get("shasum") or "") - if shasum and hashlib.sha1(data).hexdigest().lower() != shasum.lower(): - raise UpdateError("Tarball SHA-1 verification failed") + 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: @@ -198,6 +257,8 @@ LIBRARIES = ( ), ) +LIBRARIES_BY_KEY = {library.key: library for library in LIBRARIES} + def installed_version(destination: Path) -> str: version_file = destination / "VERSION" @@ -212,11 +273,14 @@ def is_complete(library: Library, destination: Path) -> bool: 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)}\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" ) @@ -228,17 +292,7 @@ def install_library( 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'}" - ) - + version, dist, _ = validate_metadata(library, metadata) tarball = request_bytes(str(dist["tarball"]), timeout) verify_tarball(tarball, dist) @@ -284,82 +338,314 @@ def selected_libraries(keys: Iterable[str]) -> list[Library]: 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, - strict: bool, ) -> bool: + target, _, _ = validate_metadata(library, metadata) 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})") + 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'} -> {latest}" + 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 {latest} ({state})") + print(f"{library.key}: downloading {target} ({state})") install_library(library, metadata, destination, timeout) - print(f"{library.key}: installed {latest}") + 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=[]) + 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 report whether an update is required") + 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() - 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): + 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, - strict=args.strict or 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 - return 0 if success else 1 + except OSError as error: + print(f"error: {error}", file=sys.stderr) + return 1 if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file