updater.py

"""Release-based auto-updater for the BC250 Toolkit Decky plugin.

Checks the latest GitHub Release, compares it to the installed version
(package.json), and — if newer — downloads the release ZIP and unpacks it over
the plugin directory, then restarts plugin_loader to reload the code.

The plugin backend runs as root and the plugin dir is root-owned, so it can
write to itself and restart the loader without sudo. Network/disk work runs in
a thread executor so the asyncio loop never blocks.
"""

import asyncio
import json
import os
import shutil
import ssl
import subprocess
import tempfile
import urllib.request
import zipfile
from pathlib import Path


def _ssl_context():
    """Le Python embarqué de plugin_loader (PyInstaller) n'a pas de bundle CA →
    urllib échoue en CERTIFICATE_VERIFY_FAILED. On pointe le bundle CA système."""
    for ca in ("/etc/pki/tls/certs/ca-bundle.crt", "/etc/ssl/certs/ca-certificates.crt", "/etc/ssl/cert.pem"):
        if os.path.exists(ca):
            try:
                return ssl.create_default_context(cafile=ca)
            except Exception:
                pass
    try:
        return ssl.create_default_context()
    except Exception:
        return None

try:
    from decky import logger  # type: ignore
except Exception:  # pragma: no cover - fallback if decky logger unavailable
    class _L:
        def info(self, m): print("[updater]", m)
        warning = error = debug = info
    logger = _L()

# --- per-plugin configuration -------------------------------------------------
GITHUB_REPO = "Necrosiak/bc250-toolkit-decky"
# -----------------------------------------------------------------------------

PLUGIN_DIR = Path(os.path.dirname(__file__))
RELEASES_API = f"https://api.github.com/repos/{GITHUB_REPO}/releases/latest"
SETTINGS_DIR = Path(os.environ.get("DECKY_PLUGIN_SETTINGS_DIR", str(PLUGIN_DIR)))
SETTINGS_FILE = SETTINGS_DIR / "updater.json"
_USER_AGENT = f"{GITHUB_REPO.split('/')[-1]}-updater"


def _parse_version(v: str) -> tuple:
    out = []
    for chunk in str(v).strip().lstrip("vV").split("."):
        digits = "".join(c for c in chunk if c.isdigit())
        out.append(int(digits) if digits else 0)
    return tuple(out) or (0,)


def get_current_version() -> str:
    try:
        data = json.loads((PLUGIN_DIR / "package.json").read_text())
        return str(data.get("version", "0.0.0"))
    except Exception as e:
        logger.warning(f"[updater] cannot read current version: {e}")
        return "0.0.0"


def is_autoupdate_enabled() -> bool:
    try:
        return bool(json.loads(SETTINGS_FILE.read_text()).get("autoupdate", True))
    except Exception:
        return True  # default ON


def set_autoupdate_enabled(enabled: bool) -> bool:
    try:
        SETTINGS_DIR.mkdir(parents=True, exist_ok=True)
        SETTINGS_FILE.write_text(json.dumps({"autoupdate": bool(enabled)}))
    except Exception as e:
        logger.warning(f"[updater] cannot persist autoupdate flag: {e}")
    return bool(enabled)


def _fetch_latest_blocking() -> dict:
    req = urllib.request.Request(
        RELEASES_API,
        headers={"User-Agent": _USER_AGENT, "Accept": "application/vnd.github+json"},
    )
    with urllib.request.urlopen(req, timeout=15, context=_ssl_context()) as resp:
        rel = json.loads(resp.read().decode("utf-8"))

    tag = rel.get("tag_name") or rel.get("name") or ""
    download_url = ""
    for asset in rel.get("assets", []):
        name = (asset.get("name") or "").lower()
        if name.endswith(".zip"):
            download_url = asset.get("browser_download_url", "")
            break
    if not download_url:
        download_url = rel.get("zipball_url", "")
    return {"tag": tag, "url": download_url, "notes": rel.get("body", "") or ""}


async def check() -> dict:
    current = get_current_version()
    try:
        loop = asyncio.get_event_loop()
        latest = await loop.run_in_executor(None, _fetch_latest_blocking)
    except Exception as e:
        logger.warning(f"[updater] check failed: {e}")
        return {"current": current, "latest": None, "update_available": False,
                "url": "", "notes": "", "error": str(e)}

    available = (
        bool(latest["tag"])
        and bool(latest["url"])
        and _parse_version(latest["tag"]) > _parse_version(current)
    )
    return {
        "current": current,
        "latest": latest["tag"],
        "update_available": available,
        "url": latest["url"],
        "notes": latest["notes"],
    }


def _content_root(extracted: Path) -> Path:
    entries = [p for p in extracted.iterdir() if not p.name.startswith("__MACOSX")]
    if len(entries) == 1 and entries[0].is_dir():
        return entries[0]
    return extracted


# Files whose absence breaks nothing if we cannot write them: documentation, a
# licence, images, and plugin.json (which Decky keeps root-owned and which only
# ever changes the name, author or flags). Everything else is code, and skipping
# a code file means shipping a half-updated plugin, i.e. a broken one.
_SKIPPABLE_NAMES = {"LICENSE", "LICENSE.md", "NOTICE", "plugin.json"}
_SKIPPABLE_SUFFIXES = {".md", ".txt", ".png", ".jpg", ".jpeg", ".svg", ".webp"}


def _writable_as_us(dst: Path) -> bool:
    """Can we write `dst` without being root?

    Decky leaves the TOP-LEVEL directory and `plugin.json` owned by root and
    chowns everything else to the host user. So there are two cases and only one
    of them works: overwriting a file that already exists and is ours needs
    write permission on the FILE, and succeeds; creating a new entry needs write
    permission on the DIRECTORY, which we do not have. Measured on a real
    install on 2026-09-13.
    """
    if dst.exists():
        return os.access(dst, os.W_OK)
    parent = dst.parent
    while not parent.exists():
        parent = parent.parent
    return os.access(parent, os.W_OK)


def _survey(root: Path, plugin_dir: Path):
    """(blocking, skippable) — what this update will not be able to write."""
    hard, soft = [], []
    for src in root.rglob("*"):
        if src.is_dir():
            continue
        rel = src.relative_to(root)
        dst = plugin_dir / rel
        if _writable_as_us(dst):
            continue
        skippable = (dst.name in _SKIPPABLE_NAMES
                     or dst.suffix.lower() in _SKIPPABLE_SUFFIXES)
        (soft if skippable else hard).append(rel)
    return hard, soft


def _apply_blocking(url: str) -> None:
    with tempfile.TemporaryDirectory() as tmp:
        tmp = Path(tmp)
        zip_path = tmp / "update.zip"
        req = urllib.request.Request(url, headers={"User-Agent": _USER_AGENT})
        with urllib.request.urlopen(req, timeout=120, context=_ssl_context()) as resp, open(zip_path, "wb") as f:
            shutil.copyfileobj(resp, f)

        extract_dir = tmp / "x"
        with zipfile.ZipFile(zip_path) as z:
            z.extractall(extract_dir)

        root = _content_root(extract_dir)

        # Survey BEFORE writing anything. The old loop wrote as it went and
        # raised on the first impossible file, leaving the plugin half updated —
        # part old code, part new — which is far worse than no update at all.
        hard, soft = _survey(root, PLUGIN_DIR)
        if hard:
            names = ", ".join(sorted(str(h) for h in hard)[:6])
            raise PermissionError(
                f"this version touches files the plugin cannot write itself "
                f"({names}) — a Decky install is required")
        if soft:
            logger.info("[updater] skipped (non-essential, not writable): "
                        + ", ".join(sorted(str(x) for x in soft)))

        skip = set(soft)
        for src in root.rglob("*"):
            rel = src.relative_to(root)
            dst = PLUGIN_DIR / rel
            if src.is_dir():
                if _writable_as_us(dst) or dst.exists():
                    dst.mkdir(parents=True, exist_ok=True)
            elif rel not in skip:
                dst.parent.mkdir(parents=True, exist_ok=True)
                _replace_file(src, dst)


def _replace_file(src: Path, dst: Path) -> None:
    """Copy src over dst via a same-directory tmp file + atomic os.replace.

    shutil.copy2 ends with a chmod on the destination, which a non-root
    process cannot do on root-owned files even when they are world-writable —
    so updates failed on root-owned installs. os.replace only needs write
    permission on the parent directory, and the replaced file then belongs to
    the current user, healing such installs one update at a time.

    But Decky only root-owns the plugin's top-level directory and plugin.json
    at install time (everything else is chowned to the host user recursively) —
    so on a real install it is the *directory* that is unwritable, while dst
    itself, when it already exists, usually is not. tmp+rename cannot work
    there, since creating the tmp file needs directory write access too, while
    overwriting an existing file's content only needs write permission on the
    file. Fall back to that. Same fix as SkullKey v1.11.2 and Steamcord #16.
    """
    tmp_dst = dst.parent / (dst.name + ".bc250tk-new")
    try:
        shutil.copyfile(src, tmp_dst)
        shutil.copymode(src, tmp_dst)  # our own file: keeps +x on binaries
        if dst.suffix in (".sh", ".py"):
            # chmod the tmp file, which is ours — chmod-ing a root-owned dst
            # is the same EPERM trap that copy2's copystat fell into.
            os.chmod(tmp_dst, 0o755)
        os.replace(tmp_dst, dst)
        return
    except PermissionError:
        try:
            os.unlink(tmp_dst)
        except OSError:
            pass
        if not dst.exists() or not os.access(dst, os.W_OK):
            raise
    except OSError:
        try:
            os.unlink(tmp_dst)
        except OSError:
            pass
        raise

    # Non-atomic (a crash mid-write leaves dst truncated), but src has already
    # been downloaded and extracted successfully before this ever runs, and the
    # alternative here is a guaranteed Permission denied.
    with open(src, "rb") as s, open(dst, "wb") as d:
        shutil.copyfileobj(s, d)
    shutil.copymode(src, dst)
    if dst.suffix in (".sh", ".py"):
        os.chmod(dst, 0o755)


async def apply(url: str) -> dict:
    """{"ok": True} ou {"ok": False, "error": "…"} — l'erreur remonte au QAM
    (avant, un échec — ex. Permission denied sur une install root-owned —
    laissait le bouton bloqué sur « installation » pour toujours)."""
    if not url:
        return {"ok": False, "error": "no url"}
    try:
        loop = asyncio.get_event_loop()
        await loop.run_in_executor(None, _apply_blocking, url)
        logger.info("[updater] update unpacked; restarting plugin_loader")
        return {"ok": True}
    except PermissionError as e:
        logger.error(f"[updater] apply failed: {e}")
        # Files are replaced via os.replace (only needs directory write
        # access), so reaching this means a *directory* is not writable.
        return {"ok": False,
                "error": f"Permission denied ({getattr(e, 'filename', '')}) — "
                         "run: sudo chown -R $(id -un) "
                         f"{PLUGIN_DIR} then retry"}
    except Exception as e:
        logger.error(f"[updater] apply failed: {e}")
        return {"ok": False, "error": str(e)}


def restart_loader() -> bool:
    """Restart Decky so the new code loads. True only if systemd accepted it.

    Two independent reasons make this fail from inside a plugin, both measured
    on 2026-09-22 (Steamcord #52 — « I click update and nothing happens »):

    1. Decky's loader is a PyInstaller bundle, so it exports its extraction dir
       on LD_LIBRARY_PATH and every plugin backend inherits it. `systemctl` then
       resolves the BUNDLED libcrypto and cannot even start:
           systemctl: /tmp/_MEIxxxxxx/libcrypto.so.3: version `OPENSSL_3.4.0'
           not found (required by …/libsystemd-shared-259.9-1.fc44.so)
       This one hits every plugin, root or not. PyInstaller keeps the caller's
       own value in LD_LIBRARY_PATH_ORIG, so restoring it hands `systemctl` a
       sane environment again.
    2. `plugin_loader` is a SYSTEM unit, so polkit answers `auth_admin` to an
       unprivileged caller — only a plugin whose `flags` contain "root" gets
       past this:
           Failed to restart plugin_loader.service: Access denied as the
           requested operation requires interactive authentication.

    `Popen` hid both: nothing waited on the child, nothing read its status,
    nothing logged. The update was written to disk and then simply never
    loaded — no progress, no message, and the new version only appeared after
    the next boot. Returning a real answer lets the caller hand the reload to
    the frontend, which can ask the loader — root — to reload this plugin alone.
    """
    env = dict(os.environ)
    orig = env.pop("LD_LIBRARY_PATH_ORIG", None)
    if orig:
        env["LD_LIBRARY_PATH"] = orig
    else:
        env.pop("LD_LIBRARY_PATH", None)
    try:
        proc = subprocess.run(
            ["systemctl", "restart", "plugin_loader"],
            capture_output=True, text=True, timeout=30, env=env,
        )
    except Exception as e:
        logger.error(f"[updater] restart failed: {e}")
        return False
    # Only reached when the restart was REFUSED: an accepted one kills this very
    # process long before subprocess.run() can return.
    msg = (proc.stderr or proc.stdout or "").strip().splitlines()
    logger.warning(
        f"[updater] loader restart refused (exit {proc.returncode}): "
        f"{msg[0] if msg else 'no output'}"
    )
    return proc.returncode == 0