"""voxip-kiosk-updater — installs new Daynivo Kiosk OS releases (runs as root). A release is a tag ``vX.Y.Z`` on the Gitea repository with two assets: ``daynivo-kiosk-os-X.Y.Z.tar.gz`` and its ``.sha256``. The updater * checks the latest release and records the result in ``update.json`` in the state directory (the web interface shows it), * downloads and verifies a newer release, unpacks it under ``/opt/daynivo-kiosk-os/`` and runs that release's ``update.sh --no-pull`` (config backup + ``install.sh --upgrade``). It never installs an older or equal version. Usage:: voxip-kiosk-updater --check # only check voxip-kiosk-updater --auto # check; install when auto_update is on (daily timer) voxip-kiosk-updater --install # check and install when newer (button in the web interface) """ from __future__ import annotations import argparse import hashlib import json import logging import re import shutil import ssl import subprocess import tarfile import tempfile import time import urllib.error import urllib.request from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Any, Callable from . import logutil from .config import atomic_write_json, load_config from .paths import Paths, default_paths, read_version log = logging.getLogger("voxip-kiosk-updater") RELEASE_API = "https://git.rothman.nl/api/v1/repos/rothman/daynivo-kiosk-os/releases/latest" ASSET_PREFIX = "daynivo-kiosk-os-" RELEASE_DIR = Path("/opt/daynivo-kiosk-os") CACHE_DIR = Path("/var/cache/voxip-kiosk") KEEP_RELEASES = 2 MAX_DOWNLOAD = 64 * 1024 * 1024 USER_AGENT = "Daynivo-Kiosk-OS/updater" _VERSION = re.compile(r"^v?(\d+)\.(\d+)\.(\d+)$") class UpdateError(Exception): pass @dataclass(frozen=True) class Release: version: str tarball_url: str checksum_url: str page_url: str def parse_version(text: str) -> tuple[int, int, int] | None: match = _VERSION.match(text.strip()) return (int(match[1]), int(match[2]), int(match[3])) if match else None def is_newer(candidate: str, current: str) -> bool: new, old = parse_version(candidate), parse_version(current.removesuffix("-dev")) return new is not None and (old is None or new > old) def http_get(url: str, limit: int = MAX_DOWNLOAD, timeout: float = 30) -> bytes: if not url.startswith("https://"): raise UpdateError(f"Alleen HTTPS-downloads: {url}") request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "application/json, */*"}) try: with urllib.request.urlopen(request, timeout=timeout, context=ssl.create_default_context()) as response: data = response.read(limit + 1) except urllib.error.HTTPError as exc: if exc.code == 404 and url == RELEASE_API: raise UpdateError("Er is nog geen release gepubliceerd") from exc raise UpdateError(f"Downloaden mislukt ({url}): HTTP {exc.code}") from exc except OSError as exc: raise UpdateError(f"Downloaden mislukt ({url}): {exc}") from exc if len(data) > limit: raise UpdateError(f"Download te groot: {url}") return data def parse_release(data: dict[str, Any]) -> Release: version = str(data.get("tag_name", "")).removeprefix("v") if parse_version(version) is None: raise UpdateError(f"Onbekend versienummer in release: {data.get('tag_name')!r}") assets = {a.get("name"): a.get("browser_download_url") for a in data.get("assets") or [] if isinstance(a, dict)} tarball = assets.get(f"{ASSET_PREFIX}{version}.tar.gz") checksum = assets.get(f"{ASSET_PREFIX}{version}.tar.gz.sha256") if not tarball or not checksum: raise UpdateError(f"Release {version} mist het installatiebestand of de checksum") return Release(version, tarball, checksum, str(data.get("html_url") or "")) def fetch_latest(get: Callable[[str], bytes] = http_get, api: str = RELEASE_API) -> Release: try: data = json.loads(get(api)) except ValueError as exc: raise UpdateError("Ongeldig antwoord van de releaseserver") from exc if not isinstance(data, dict): raise UpdateError("Ongeldig antwoord van de releaseserver") return parse_release(data) def read_state(paths: Paths) -> dict[str, Any]: try: data = json.loads(paths.update_state_file.read_text(encoding="utf-8")) except (OSError, ValueError): return {} return data if isinstance(data, dict) else {} def safe_extract(archive: Path, target: Path) -> None: """Unpack only regular files and directories that stay inside ``target``.""" with tarfile.open(archive, "r:gz") as tar: members = tar.getmembers() for member in members: path = PurePosixPath(member.name) if path.is_absolute() or ".." in path.parts or not (member.isfile() or member.isdir()): raise UpdateError(f"Onveilig bestand in release: {member.name}") tar.extractall(target, members=members) # noqa: S202 - members checked above class Updater: def __init__(self, paths: Paths | None = None, get: Callable[[str], bytes] = http_get, run: Callable[..., Any] = subprocess.run, release_dir: Path = RELEASE_DIR, cache_dir: Path = CACHE_DIR, now: Callable[[], float] = time.time): self.paths = paths or default_paths() self.get = get self.run = run self.release_dir = release_dir self.cache_dir = cache_dir self.now = now def _write_state(self, **values: Any) -> dict[str, Any]: state = {**read_state(self.paths), **values} try: atomic_write_json(self.paths.update_state_file, state, mode=0o644, owner=(0, 0)) except PermissionError: atomic_write_json(self.paths.update_state_file, state, mode=0o644) except OSError as exc: log.warning("Kon updatestatus niet opslaan: %s", exc) return state def check(self) -> tuple[dict[str, Any], Release | None]: current = read_version(self.paths) try: release = fetch_latest(self.get) except UpdateError as exc: log.warning("Controleren op updates mislukt: %s", exc) return self._write_state(checked_at=int(self.now()), current=current, check_error=str(exc)), None available = is_newer(release.version, current) state = self._write_state(checked_at=int(self.now()), current=current, latest=release.version, available=available, release_url=release.page_url, check_error=None) if available: log.info("Nieuwe versie beschikbaar: %s (geïnstalleerd: %s)", release.version, current) return state, release if available else None def install(self, release: Release) -> dict[str, Any]: current = read_version(self.paths) if not is_newer(release.version, current): raise UpdateError(f"Versie {release.version} is niet nieuwer dan {current}") self._write_state(status="updating", target=release.version, started_at=int(self.now()), install_error=None) log.info("Bijwerken naar %s", release.version) try: folder = self._download(release) proc = self.run([str(folder / "update.sh"), "--no-pull"], capture_output=True, text=True, timeout=3600) if proc.returncode != 0: tail = (proc.stderr or proc.stdout or "").strip().splitlines()[-5:] raise UpdateError("Installatie mislukt: " + " / ".join(tail)) except (UpdateError, OSError, subprocess.SubprocessError) as exc: log.error("Bijwerken naar %s mislukt: %s", release.version, exc) return self._write_state(status="failed", finished_at=int(self.now()), install_error=str(exc)) self._prune(keep=release.version) log.info("Bijgewerkt naar %s", release.version) return self._write_state(status="done", finished_at=int(self.now()), current=release.version, available=False, install_error=None) def _download(self, release: Release) -> Path: tarball = self.get(release.tarball_url) expected = self.get(release.checksum_url, limit=4096).decode("ascii", "replace").split() actual = hashlib.sha256(tarball).hexdigest() if not expected or expected[0].lower() != actual: raise UpdateError("Checksum klopt niet; de download wordt niet geïnstalleerd") self.cache_dir.mkdir(parents=True, exist_ok=True) self.release_dir.mkdir(parents=True, exist_ok=True) with tempfile.TemporaryDirectory(dir=self.cache_dir) as tmp: archive = Path(tmp) / "release.tar.gz" archive.write_bytes(tarball) unpacked = Path(tmp) / "unpacked" safe_extract(archive, unpacked) source = unpacked / f"{ASSET_PREFIX}{release.version}" found = (source / "VERSION").read_text(encoding="utf-8").strip() if (source / "VERSION").exists() else None if found != release.version: raise UpdateError(f"Release bevat versie {found!r} in plaats van {release.version}") target = self.release_dir / release.version shutil.rmtree(target, ignore_errors=True) shutil.move(str(source), target) return target def _prune(self, keep: str) -> None: versions = sorted((p for p in self.release_dir.iterdir() if p.is_dir() and parse_version(p.name)), key=lambda p: parse_version(p.name), reverse=True) for old in versions[KEEP_RELEASES:]: if old.name != keep: shutil.rmtree(old, ignore_errors=True) def main(argv: list[str] | None = None, updater: Updater | None = None) -> int: parser = argparse.ArgumentParser(prog="voxip-kiosk-updater", description="Daynivo Kiosk OS bijwerken.") mode = parser.add_mutually_exclusive_group(required=True) mode.add_argument("--check", action="store_true", help="alleen controleren") mode.add_argument("--auto", action="store_true", help="controleren en bijwerken als automatisch bijwerken aan staat") mode.add_argument("--install", action="store_true", help="controleren en een nieuwere versie installeren") args = parser.parse_args(argv) logutil.setup() updater = updater or Updater() state, release = updater.check() if args.check or release is None: print(json.dumps(state, ensure_ascii=False)) return 1 if state.get("check_error") else 0 if args.auto: if not load_config(updater.paths.config_file)["auto_update"]: log.info("Automatisch bijwerken staat uit; %s wordt niet geïnstalleerd", release.version) return 0 if (updater.paths.health_run_dir / "maintenance").exists(): log.info("Onderhoudsmodus actief; bijwerken overgeslagen") return 0 state = updater.install(release) print(json.dumps(state, ensure_ascii=False)) return 0 if state.get("status") == "done" else 1