#!/usr/bin/env python3 """Offline checksum-pinned install; never writes a repository or credentials.""" import argparse import hashlib import json import os import pathlib import re import shutil import subprocess import sys import tarfile import tempfile def require(condition: bool, message: str) -> None: if not condition: raise ValueError(message) def file_digest(path: pathlib.Path) -> str: with path.open("rb") as source: return hashlib.file_digest(source, "sha256").hexdigest() def check_node() -> None: binary = shutil.which("node") require(binary is not None, "NODE_22_REQUIRED") result = subprocess.run( # noqa: S603 — fixed version probe on the selected local runtime. [binary, "--version"], capture_output=True, text=True, timeout=10, check=False, ) match = re.fullmatch(r"v([0-9]+)\.[0-9]+\.[0-9]+", result.stdout.strip()) require( result.returncode == 0 and match is not None and int(match.group(1)) >= 22, "NODE_22_REQUIRED", ) def validate_prefix(prefix: pathlib.Path) -> pathlib.Path: require( prefix.stat().st_uid == os.getuid() and prefix.stat().st_mode & 0o022 == 0, "INSTALL_PREFIX_NOT_PRIVATE_TO_OWNER", ) for name in ("versions", "bin"): require(not (prefix / name).is_symlink(), f"{name.upper()}_DIRECTORY_SYMLINK") launcher = prefix / "bin/factoryos-runner" managed = ( launcher.is_symlink() and os.readlink(launcher) == "../current/bin/factoryos-runner.mjs" ) require( not launcher.exists() and not launcher.is_symlink() or managed, "EXISTING_LAUNCHER_PRESERVED", ) current = prefix / "current" require(not current.exists() or current.is_symlink(), "CURRENT_IS_NOT_MANAGED_LINK") pending = prefix / ".current-new" require(not pending.exists() and not pending.is_symlink(), "ACTIVATION_PENDING") return launcher def extract(archive: pathlib.Path, stage: pathlib.Path) -> set[str]: names: set[str] = set() with tarfile.open(archive, "r:gz") as source: members = source.getmembers() require( len(members) <= 20000 and sum(m.size for m in members) <= 256 * 1024 * 1024, "ARCHIVE_TOO_LARGE", ) for member in members: path = pathlib.PurePosixPath(member.name) require( member.isfile() and not path.is_absolute() and ".." not in path.parts and str(path) == member.name and member.name not in names, "UNSAFE_ARCHIVE_MEMBER", ) names.add(member.name) destination = stage / member.name destination.parent.mkdir(parents=True, exist_ok=True) data = source.extractfile(member) require(data is not None, "UNREADABLE_ARCHIVE_MEMBER") with data, destination.open("xb") as target: shutil.copyfileobj(data, target) destination.chmod(0o755 if member.mode & 0o111 else 0o644) return names def verify(stage: pathlib.Path, names: set[str]) -> str: manifest = json.loads((stage / "manifest.json").read_text()) require( isinstance(manifest, dict) and set(manifest) == names - {"manifest.json"}, "MANIFEST_MEMBERS_DIFFER", ) for name, expected in manifest.items(): require(file_digest(stage / name) == expected, "MANIFEST_DIGEST_MISMATCH") release = json.loads((stage / "release.json").read_text()) require( isinstance(release, dict) and release.get("schema") == "factoryos-runner-release:v1", "UNSUPPORTED_RELEASE", ) version = release.get("version") require( isinstance(version, str) and re.fullmatch(r"[0-9]+\.[0-9]+\.[0-9]+(?:-[a-zA-Z0-9.-]+)?", version) is not None, "UNSUPPORTED_RELEASE", ) return version def publish( stage: pathlib.Path, prefix: pathlib.Path, version: str, activate: bool, launcher: pathlib.Path, ) -> pathlib.Path: destination = prefix / "versions" / version destination.parent.mkdir(mode=0o700, exist_ok=True) require(not destination.exists() and not destination.is_symlink(), "VERSION_ALREADY_INSTALLED") # Updates never rewrite a version directory; existing processes keep their paths. os.rename(stage, destination) stage.mkdir() # TemporaryDirectory retains only its original staging path. if activate: pending = prefix / ".current-new" pending.symlink_to(pathlib.Path("versions") / version) os.replace(pending, prefix / "current") launcher.parent.mkdir(mode=0o700, exist_ok=True) if not launcher.is_symlink(): launcher.symlink_to("../current/bin/factoryos-runner.mjs") return destination def install( archive: pathlib.Path, digest: str, prefix: pathlib.Path, activate: bool, ) -> pathlib.Path: require(sys.version_info >= (3, 11), "PYTHON_311_REQUIRED") require(not prefix.is_symlink(), "INSTALL_PREFIX_SYMLINK") require(re.fullmatch(r"[a-f0-9]{64}", digest) is not None, "SHA256_REQUIRED") require(file_digest(archive) == digest, "ARCHIVE_DIGEST_MISMATCH") check_node() prefix.mkdir(mode=0o700, parents=True, exist_ok=True) prefix = prefix.resolve() launcher = validate_prefix(prefix) lock = prefix / ".install-lock" try: lock.mkdir(mode=0o700) except FileExistsError as error: raise ValueError("INSTALLATION_LOCKED") from error try: with tempfile.TemporaryDirectory(prefix=".install-", dir=prefix) as temporary: stage = pathlib.Path(temporary) version = verify(stage, extract(archive, stage)) return publish(stage, prefix, version, activate, launcher) finally: lock.rmdir() def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("archive", type=pathlib.Path) parser.add_argument( "--sha256", required=True, help="expected digest from trusted release record", ) parser.add_argument( "--prefix", type=pathlib.Path, default=pathlib.Path.home() / ".local/lib/factoryos-runner", ) parser.add_argument("--activate", action="store_true", help="explicitly select this version") args = parser.parse_args() destination = install(args.archive, args.sha256, args.prefix.absolute(), args.activate) result = { "installed": str(destination), "activated": args.activate, "command": str(args.prefix.absolute() / "bin/factoryos-runner"), } sys.stdout.write(json.dumps(result) + "\n") if __name__ == "__main__": try: main() except ( ValueError, OSError, KeyError, TypeError, tarfile.TarError, subprocess.TimeoutExpired, ) as error: code = str(error) if re.fullmatch(r"[A-Z0-9_]+", str(error)) else "INSTALL_FAILED" sys.stderr.write(json.dumps({"success": False, "code": code}) + "\n") sys.exit(1)