gru/rustpad
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

new functions and fixes

Browse Source
This commit is contained in:
Mateusz Gruszczyński
2026-08-05 22:24:54 +02:00
parent e842b26978
commit 19d481fe72
3 changed files with 374 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Dockerfile
+3 -2
View File
@@ -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
scripts/browser-libs.lock.json
+21
View File
@@ -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
}
scripts/update_browser_libs.py
+350 -64
View File
@@ -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())