Cachy-Predator-Keyboard/predator_control/updater.py

"""Self-updater: checks the Gitea releases for a newer pacman package and installs it.

The release must carry a `predator-control-<ver>-<rel>-any.pkg.tar.zst` asset and its
SHA-256 in the release notes. Downloads are verified before `pacman -U` runs.
An optional API token (for private repositories) is read from
/etc/predator-control/update-token.
"""

import hashlib
import json
import os
import re
import shutil
import subprocess
import time
import urllib.request

from . import CONFIG_DIR, __version__

API = "https://git.mrblake.cc/api/v1/repos/MrBlake/Cachy-Predator-Keyboard"
CACHE_DIR = "/var/cache/predator-control"
TOKEN_PATH = CONFIG_DIR + "/update-token"
PKG_RE = re.compile(r"^predator-control-[\w.+]+-\d+-any\.pkg\.tar\.zst$")
UNIT = "predator-control-update"


class UpdateError(RuntimeError):
    pass


def parse_version(v):
    return tuple(int(x) for x in re.findall(r"\d+", v.lstrip("v"))[:3])


def _request(url, binary=False):
    headers = {"Accept": "application/json", "User-Agent": "predator-control/" + __version__}
    try:
        with open(TOKEN_PATH) as f:
            token = f.read().strip()
        if token:
            headers["Authorization"] = "token " + token
    except OSError:
        pass
    try:
        with urllib.request.urlopen(urllib.request.Request(url, headers=headers), timeout=30) as r:
            data = r.read()
    except OSError as e:
        raise UpdateError("cannot reach the update server: %s" % e) from e
    return data if binary else json.loads(data)


def installed_as_package():
    """True if the app was installed through pacman (only then can it update itself)."""
    if not shutil.which("pacman"):
        return False
    return subprocess.run(["pacman", "-Qq", "predator-control"], capture_output=True).returncode == 0


def check():
    """Return information about the latest release."""
    releases = [r for r in _request(API + "/releases?limit=50")
                if not r.get("draft") and not r.get("prerelease") and parse_version(r.get("tag_name", ""))]
    if not releases:
        raise UpdateError("no releases found")
    rel = max(releases, key=lambda r: parse_version(r["tag_name"]))
    latest = rel.get("tag_name", "").lstrip("v")
    asset = next((a for a in rel.get("assets", []) if PKG_RE.match(a.get("name", ""))), None)
    sha = None
    if asset:
        m = re.search(re.escape(asset["name"]) + r"\W+([0-9a-f]{64})", rel.get("body", ""))
        sha = m.group(1) if m else None
    return {
        "current": __version__,
        "latest": latest,
        "available": parse_version(latest) > parse_version(__version__),
        "notes": rel.get("body", ""),
        "url": rel.get("html_url"),
        "asset": asset["name"] if asset else None,
        "asset_url": asset["browser_download_url"] if asset else None,
        "sha256": sha,
        "can_install": installed_as_package() and bool(asset and sha),
        "checked": time.time(),
    }


def install(info):
    """Download, verify and install the package in a separate systemd unit.

    pacman's post_upgrade restarts predatord, so the installation must not run
    inside the daemon's own process.
    """
    if not info.get("available"):
        raise UpdateError("already up to date")
    if not installed_as_package():
        raise UpdateError("not installed via pacman – update with git pull && sudo ./install.sh")
    if not info.get("asset_url") or not info.get("sha256"):
        raise UpdateError("release has no verifiable package")
    os.makedirs(CACHE_DIR, exist_ok=True)
    path = os.path.join(CACHE_DIR, info["asset"])
    data = _request(info["asset_url"], binary=True)
    digest = hashlib.sha256(data).hexdigest()
    if digest != info["sha256"]:
        raise UpdateError("checksum mismatch – download rejected")
    with open(path, "wb") as f:
        f.write(data)
    subprocess.run(["systemctl", "reset-failed", UNIT + ".service"], capture_output=True)
    r = subprocess.run(["systemd-run", "--unit=" + UNIT, "--description=Predator Control update",
                        "pacman", "-U", "--noconfirm", "--needed", path],
                       capture_output=True, text=True)
    if r.returncode != 0:
        raise UpdateError("could not start the update: %s" % r.stderr.strip())
    return {"started": True, "package": info["asset"]}


def status():
    """State of the installer unit: 'running', 'failed' or 'idle'."""
    r = subprocess.run(["systemctl", "is-active", UNIT + ".service"], capture_output=True, text=True)
    s = r.stdout.strip()
    if s in ("active", "activating"):
        return "running"
    if s == "failed":
        return "failed"
    return "idle"