"""CachyOS Update - Decky plugin backend. Runs as root (plugin.json declares the "root" flag) and drives pacman, yay, flatpak and fwupdmgr non-interactively, streaming every output line to the frontend. All user-facing wording lives in the frontend (src/i18n.ts). This module only ever returns stable identifiers - phase ids, hint ids, reason ids - so the UI can render them in the user's language. Raw command output is passed through untranslated on purpose: it is the tool's own output. This file is also runnable standalone for development on a machine without Decky installed: python3 main.py --selftest python3 main.py --check python3 main.py --update --dry-run sudo python3 main.py --fix-lock """ import asyncio import json import os import re import shutil import sys import time from collections import deque from pathlib import Path try: import decky # type: ignore except ImportError: # standalone / development decky = None # -------------------------------------------------------------------------- # decky stub for standalone runs # -------------------------------------------------------------------------- if decky is None: class _StubLogger: def _p(self, level, msg, *a): print(f"[{level}] {msg % a if a else msg}", file=sys.stderr) def debug(self, msg, *a): self._p("debug", msg, *a) def info(self, msg, *a): self._p("info", msg, *a) def warning(self, msg, *a): self._p("warn", msg, *a) def error(self, msg, *a): self._p("error", msg, *a) def exception(self, msg, *a): self._p("error", msg, *a) class _StubDecky: """Mimics the subset of the decky module that this plugin uses.""" def __init__(self): base = Path(os.environ.get("TMPDIR", "/tmp")) / "decky-cachyos-update-dev" (base / "settings").mkdir(parents=True, exist_ok=True) (base / "runtime").mkdir(parents=True, exist_ok=True) (base / "logs").mkdir(parents=True, exist_ok=True) self.DECKY_PLUGIN_SETTINGS_DIR = str(base / "settings") self.DECKY_PLUGIN_RUNTIME_DIR = str(base / "runtime") self.DECKY_PLUGIN_LOG_DIR = str(base / "logs") self.DECKY_USER = os.environ.get("USER", "root") self.DECKY_USER_HOME = os.path.expanduser("~") self.DECKY_PLUGIN_VERSION = "dev" self.logger = _StubLogger() async def emit(self, event, *args): if event == "cachyos_update_log": print(args[0]) else: print(f"<{event}> {args}", file=sys.stderr) decky = _StubDecky() # type: ignore STANDALONE = True else: STANDALONE = False # -------------------------------------------------------------------------- # constants # -------------------------------------------------------------------------- LOG_MAX_LINES = 500 PACMAN_LOCK = "/var/lib/pacman/db.lck" # How long a lock nobody has open has to have been lying around before it # counts as abandoned. libalpm creates the file and its descriptor in the same # atomic open(), so there is no window in which a live lock looks unowned - # this is pure insurance against a /proc scan that raced with a process start. LOCK_STALE_AFTER = 5.0 # sudo skips files in sudoers.d whose name contains a "." or ends in "~", # so the name deliberately has neither. SUDOERS_FILE = "/etc/sudoers.d/50-decky-cachyos-update" # Weight of each phase in the progress bar, roughly proportional to how long # each step usually takes on a handheld. PHASE_WEIGHTS = { "keyring": 5, "pacman": 40, "aur": 30, "flatpak_system": 12, "flatpak_user": 5, "fwupd": 8, } DEFAULT_SETTINGS = { "enable_aur": True, "enable_flatpak": True, "enable_fwupd": False, "auto_check": True, "auto_install": False, "check_interval_hours": 6, "notify_on_updates": True, # Only notify once at least this many packages are waiting. "notify_threshold": 50, } # Packages whose upgrade means the machine should be restarted. Mirrors the # trigger list of /usr/share/libalpm/hooks/cachyos-reboot-required.hook. REBOOT_TRIGGER_PATTERNS = [ re.compile(r"^linux(-.*)?$"), re.compile(r"^systemd(-.*)?$"), re.compile(r"^mesa$"), re.compile(r"^nvidia(-.*)?$"), re.compile(r"^wayland$"), re.compile(r"^xorg-server(-.*)?$"), re.compile(r"^mkinitcpio(-.*)?$"), re.compile(r"^dracut(-.*)?$"), re.compile(r"^(amd|intel)-ucode$"), re.compile(r"^linux-firmware(-.*)?$"), re.compile(r"^glibc$"), ] # Output markers that mean "this needs a human", mapped to a hint id that the # frontend turns into localized advice. Order matters: first match wins. ERROR_HINTS = [ # Only reachable while another package manager genuinely holds the lock - # an abandoned one is removed before the run even starts. ("unable to lock database", "db_locked"), ("failed to init transaction", "db_locked"), ("unresolvable package conflicts", "pkg_conflict"), ("conflicting files", "file_conflict"), ("exists in filesystem", "file_conflict"), ("unresolvable dependencies", "unresolvable_deps"), # Only reached when holding the package back did not help either. ("breaks dependency", "dep_break"), ("could not satisfy dependencies", "dep_break"), ("invalid or corrupted package", "corrupt_package"), ("signature from", "bad_signature"), ("failed to synchronize", "sync_failed"), ("could not resolve host", "no_network"), ("failed to obtain auth", "fwupd_auth"), ("not authorized", "polkit_denied"), ("ac power required", "ac_power_required"), ("battery level is too low", "battery_too_low"), ("not found (required by", "library_conflict"), # Only reachable if the fallback (yay as root, sandboxed) ever runs: # /var/cache/private is deliberately unreachable for unprivileged users. ("/var/cache/private", "aur_sandbox"), # makepkg runs inside systemd-run, which does not inherit the LC_ALL we # set, so its messages arrive in the system language. Match both, and keep # a bare "pgp" as a last resort for any other locale. ("one or more pgp signatures could not be verified", "aur_pgp"), ("pgp-signaturen konnten nicht", "aur_pgp"), ("unknown public key", "aur_pgp"), ("unbekannter \u00f6ffentlicher schl\u00fcssel", "aur_pgp"), ("pgp", "aur_pgp"), ] PROGRESS_RE = re.compile(r"\(\s*(\d+)\s*/\s*(\d+)\s*\)") UPGRADING_RE = re.compile(r"upgrading\s+([a-zA-Z0-9@._+-]+)") # ":: installing libpcap (1.11.0-1.1) breaks dependency 'libpcap=1.10.6' # required by lib32-libpcap" - the two packages have to move together and one # of them is not ready. LC_ALL is pinned to C.UTF-8, so this wording is the # only one that can appear. DEP_BREAK_RE = re.compile( r"^::\s+(?:installing|upgrading)\s+(\S+?)(?:\s+\([^)]*\))?\s+" r"breaks dependency\s+'[^']+'\s+required by\s+(\S+)" ) # How many packages may be held back before the situation stops looking like # a repository that is a few hours behind and starts looking like a broken # system the user needs to know about. MAX_HELD_PACKAGES = 10 # yay prints its early warnings through fallbackLog, before --color never is # parsed, so colour codes show up even with colour disabled. ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]") # "package 1.0-1 -> 1.1-1 [2h5m]". Matching loosely on "->" is not enough: # yay formats its warnings as "-> message", which would be counted as updates. AUR_UPDATE_RE = re.compile(r"^\S+\s+\S+\s+->\s+\S+") # With --noprogressbar pacman prints nothing at all while downloading, which # on a slow connection means minutes of silence. Knowing the size up front is # what makes that wait understandable. DOWNLOAD_SIZE_RE = re.compile(r"Total Download Size:\s*([\d.]+)\s*MiB") # -------------------------------------------------------------------------- # state # -------------------------------------------------------------------------- class State: def __init__(self): self.status = "idle" # idle | checking | updating | done | error self.phase = "" self.progress = 0.0 self.update_started = 0.0 self.download_mib = 0.0 self.downloading = False self.log = deque(maxlen=LOG_MAX_LINES) self.counts = {"pacman": 0, "aur": 0, "flatpak": 0, "fwupd": 0} self.details = {"pacman": [], "aur": [], "flatpak": [], "fwupd": []} self.last_check = 0.0 self.last_update = 0.0 self.reboot_required = False self.failed_phases = [] self.failed_packages = [] self.held_packages = [] self.hint_id = "" self.warnings = [] self.error_text = "" # only for unexpected exceptions def to_dict(self): return { "status": self.status, "phase": self.phase, "progress": self.progress, "update_started": self.update_started, "download_mib": self.download_mib, "downloading": self.downloading, "log": list(self.log), "counts": self.counts, "details": self.details, "total": sum(self.counts.values()), "last_check": self.last_check, "last_update": self.last_update, "reboot_required": self.reboot_required, "failed_phases": self.failed_phases, "failed_packages": self.failed_packages, "held_packages": self.held_packages, "hint_id": self.hint_id, "warnings": self.warnings, "error_text": self.error_text, } # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def _which(binary): return shutil.which(binary) def _base_env(): """Deterministic, non-interactive environment for every child process. Decky's plugin_loader is a PyInstaller one-file bundle. It unpacks its own copies of libssl, libcrypto and friends into /tmp/_MEIxxxxxx and points LD_LIBRARY_PATH at that directory. Every system binary started from here would then load those instead of the real ones, and pacman dies with "version `OPENSSL_3.2.0' not found (required by libcurl.so.4)". PyInstaller keeps the original values in *_ORIG, so they can be restored. """ env = dict(os.environ) for name in ("LD_LIBRARY_PATH", "LD_PRELOAD"): original = env.pop(f"{name}_ORIG", None) if original: env[name] = original else: env.pop(name, None) # Catch anything else still pointing into the bundle (PYTHONHOME, # SSL_CERT_FILE, ...) - the temp dir name is the reliable marker. for key in [k for k, v in env.items() if isinstance(v, str) and "/_MEI" in v]: env.pop(key, None) env.pop("_MEIPASS2", None) # C.UTF-8 keeps pacman's error strings parseable regardless of the user's # locale - the ERROR_HINTS matching below relies on the English wording. env["LC_ALL"] = "C.UTF-8" env["LANG"] = "C.UTF-8" env["TERM"] = "dumb" env["PYTHONUNBUFFERED"] = "1" # Suppresses flatpak's ANSI progress bars and OSC escape sequences, which # would otherwise end up as garbage in the log panel. env["FLATPAK_FANCY_OUTPUT"] = "0" return env def _yay_env(): """Environment for yay when it runs as root. yay de-elevates makepkg/git/gpg in one of two ways (pkg/settings/exe/cmd_builder.go, deElevateCommand): - With SUDO_USER set it sets cmd.SysProcAttr.Credential to that user. **This path is broken in yay 13.0.1**: OSRunner.Show() and Capture() in pkg/settings/exe/exec.go then do `cmd.SysProcAttr = &syscall.SysProcAttr{ Pdeathsig: ...}`, throwing the Credential away. makepkg ends up running as root and aborts with "Running makepkg as root is not allowed" (exit 10). Do not set SUDO_USER here - it makes every AUR build fail. - Without it, yay wraps each step in `systemd-run -p DynamicUser=yes`, which replaces the command rather than SysProcAttr and therefore survives. That is the working path, so keep SUDO_USER unset. It also means the build directory must stay /var/cache/yay, which systemd creates via CacheDirectory=yay. """ env = _base_env() for key in ("SUDO_USER", "DOAS_USER", "SUDO_UID", "SUDO_GID"): env.pop(key, None) return env def _is_root(): return os.geteuid() == 0 def _desktop_user(): """The real desktop user, or ("", "") if we cannot determine it.""" user = getattr(decky, "DECKY_USER", "") or "" home = getattr(decky, "DECKY_USER_HOME", "") or "" return (user, home) if user and user != "root" and home else ("", "") def _can_run_as_user(): """True if work can be performed under the desktop user's identity. Either we already are that user (development), or we are root and can drop to them via runuser. """ user, _ = _desktop_user() if not user: return False return not _is_root() or bool(_which("runuser")) def _user_cmd(cmd, extra_env=None): """Return (command, env) that runs `cmd` as the desktop user. Anything that only reads - checkupdates, yay -Qua, flatpak --user - must not run as root. yay in particular resolves its cache to /var/cache/yay when root and deliberately does not create it (it relies on systemd-run's CacheDirectory), so a plain query as root fails outright. `runuser` sets HOME/USER/LOGNAME but inherits everything else. A leaked XDG_DATA_HOME or FLATPAK_USER_DIR would silently point the tools at the wrong tree, so the environment is rebuilt from scratch with `env -i` rather than scrubbed. """ extra_env = extra_env or {} user, home = _desktop_user() if not _is_root() or not user or not _which("runuser"): env = _base_env() env.update(extra_env) return cmd, env env_args = [ f"HOME={home}", f"USER={user}", f"LOGNAME={user}", "PATH=/usr/local/bin:/usr/bin:/bin", "LC_ALL=C.UTF-8", "LANG=C.UTF-8", "FLATPAK_FANCY_OUTPUT=0", ] + [f"{k}={v}" for k, v in extra_env.items()] return ["runuser", "-u", user, "--", "env", "-i"] + env_args + cmd, _base_env() # -------------------------------------------------------------------------- # pacman database lock # # An interrupted pacman - a killed pamac, a suspend, a reboot in the middle of # a transaction - leaves /var/lib/pacman/db.lck behind. Nothing in the system # ever cleans that up, so every later run dies with "unable to lock database" # until someone deletes the file by hand. Since the plugin is what people use # to update, it has to recognise and clear that itself. # -------------------------------------------------------------------------- def _proc_name(pid): try: with open(f"/proc/{pid}/comm", encoding="utf-8", errors="replace") as fh: return fh.read().strip() except OSError: return "" def _lock_holder(): """(pid, name) of the process that has the pacman lock open, or None. libalpm creates db.lck with O_CREAT|O_EXCL and keeps the descriptor open until the transaction ends, so an open descriptor pointing at the file is proof that the lock is live - and the absence of one is proof that it is not, no matter how the owner died. That is far more reliable than looking for process names: pamac's daemon and yay's helpers come and go, and the file carries no pid. Reading other processes' descriptors requires root, which the backend has. """ targets = {PACMAN_LOCK, os.path.realpath(PACMAN_LOCK)} # A holder that still has the deleted file open keeps blocking libalpm's # own O_EXCL create, so it counts just the same. targets |= {f"{path} (deleted)" for path in tuple(targets)} try: entries = list(os.scandir("/proc")) except OSError: return None for entry in entries: if not entry.name.isdigit(): continue fd_dir = f"/proc/{entry.name}/fd" try: fds = os.listdir(fd_dir) except OSError: # Exited between the two calls, or belongs to another user. continue for fd in fds: try: if os.readlink(f"{fd_dir}/{fd}") in targets: return int(entry.name), _proc_name(entry.name) or "?" except OSError: continue return None def _lock_status(): """What the pacman database lock is doing right now. "held" means a live package manager owns it and we have to wait; "stale" means the file is an orphan and removing it is safe. """ empty = { "present": False, "held": False, "stale": False, "holder": "", "holder_pid": 0, "age": 0.0, "inode": 0, } try: info = os.stat(PACMAN_LOCK) except OSError: return empty status = dict( empty, present=True, # The file stays empty for its whole lifetime, so its mtime is the # moment the transaction started. age=max(0.0, time.time() - info.st_mtime), inode=info.st_ino, ) holder = _lock_holder() if holder: return dict(status, held=True, holder_pid=holder[0], holder=holder[1]) # Without root the scan only sees our own descriptors, so "nobody has it # open" would be a guess - and a wrong guess would delete a live lock. # Assume it is held instead; in production the backend runs as root. if not _is_root(): return dict(status, held=True) return dict(status, stale=status["age"] >= LOCK_STALE_AFTER) def _clear_stale_lock(): """Remove an abandoned pacman lock. Returns its status dict, or None. Deleting the file is exactly what pacman's own error message asks for, and it is safe here because no process has the database open - there is no transaction left that could be corrupted. """ status = _lock_status() if not status["stale"]: return None try: # Anything that grabbed the lock in the meantime had to reuse this # very file (O_EXCL fails while it exists), so an inode that no longer # matches means someone else already cleaned up and started fresh. if os.stat(PACMAN_LOCK).st_ino != status["inode"]: return None os.remove(PACMAN_LOCK) except OSError as exc: decky.logger.error("Could not remove stale pacman lock: %s", exc) return None decky.logger.warning( "Removed stale pacman lock %s (%d s old, no process had it open)", PACMAN_LOCK, int(status["age"]), ) return status _installed_cache = None def _installed(pkg): """True if a package is installed, without shelling out to pacman. The local database has thousands of entries, so the name set is built once and reused; it only changes across an update run, which invalidates it. """ global _installed_cache if _installed_cache is None: local = Path("/var/lib/pacman/local") try: _installed_cache = { p.name.rsplit("-", 2)[0] for p in local.iterdir() if p.is_dir() } except OSError: _installed_cache = set() return pkg in _installed_cache def _invalidate_installed_cache(): global _installed_cache _installed_cache = None def _dependency_breaks(lines): """(blocker, dependant) pairs pacman aborted the transaction over. A package with a strict "=" dependency on a library has to be upgraded in the same transaction as that library. When it cannot be - because Arch's multilib repository trails the others, or because the dependant comes from the AUR and has to be rebuilt first - pacman refuses the whole transaction, which would mean no updates at all until someone sorts the pair out by hand. """ pairs = [] for line in lines: match = DEP_BREAK_RE.match(line.strip()) if match: pair = (match.group(1), match.group(2)) if pair not in pairs: pairs.append(pair) return pairs # -------------------------------------------------------------------------- # plugin # -------------------------------------------------------------------------- class Plugin: def __init__(self): self.state = State() self.settings = dict(DEFAULT_SETTINGS) self.lock = asyncio.Lock() self._auto_task = None self._dry_run = False self._log_file = None # -- lifecycle --------------------------------------------------------- async def _main(self): self._load_settings() self._restore_state() decky.logger.info("CachyOS Update backend started (root=%s)", _is_root()) if not _is_root(): decky.logger.error( 'Backend is NOT running as root - plugin.json is missing the ' '"root" flag. Updates will fail.' ) # A crash or forced restart could have left the rule behind. self._revoke_pacman() # Same for the database lock: if the machine went down mid-transaction, # this is the first moment anything can notice and repair it. No # update log is open yet and no frontend is listening, so this goes to # the plugin log only. _clear_stale_lock() self._auto_task = asyncio.create_task(self._auto_check_loop()) async def _unload(self): if self._auto_task: self._auto_task.cancel() self._revoke_pacman() # Harmless while an update is still running - a lock its pacman holds # open is not stale, so it stays. _clear_stale_lock() self._save_state() decky.logger.info("CachyOS Update backend stopped") # -- persistence ------------------------------------------------------- @property def _settings_file(self): return Path(decky.DECKY_PLUGIN_SETTINGS_DIR) / "settings.json" @property def _state_file(self): return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "state.json" def _load_settings(self): try: if self._settings_file.exists(): stored = json.loads(self._settings_file.read_text()) for key in DEFAULT_SETTINGS: if key in stored: self.settings[key] = stored[key] except Exception as exc: decky.logger.warning("Could not read settings: %s", exc) def _save_settings(self): try: self._settings_file.parent.mkdir(parents=True, exist_ok=True) self._settings_file.write_text(json.dumps(self.settings, indent=2)) except Exception as exc: decky.logger.warning("Could not save settings: %s", exc) def _restore_state(self): """Carry the last check result across plugin restarts.""" try: if not self._state_file.exists(): return stored = json.loads(self._state_file.read_text()) self.state.counts = stored.get("counts", self.state.counts) self.state.details = stored.get("details", self.state.details) self.state.last_check = stored.get("last_check", 0.0) self.state.last_update = stored.get("last_update", 0.0) except Exception as exc: decky.logger.warning("Could not read stored state: %s", exc) def _save_state(self): try: self._state_file.parent.mkdir(parents=True, exist_ok=True) self._state_file.write_text( json.dumps( { "counts": self.state.counts, "details": self.state.details, "last_check": self.state.last_check, "last_update": self.state.last_update, }, indent=2, ) ) except Exception as exc: decky.logger.warning("Could not save state: %s", exc) # -- emitting ---------------------------------------------------------- @property def _update_log_path(self): base = getattr(decky, "DECKY_PLUGIN_LOG_DIR", None) or decky.DECKY_PLUGIN_RUNTIME_DIR return Path(base) / "update.log" def _open_update_log(self): """Keep the full output of one run on disk. The in-memory ring buffer only holds the last few hundred lines and is gone as soon as the plugin reloads - which is exactly when someone wants to look at why a build failed. """ self._close_update_log() try: self._update_log_path.parent.mkdir(parents=True, exist_ok=True) self._log_file = self._update_log_path.open("w", encoding="utf-8") except OSError as exc: decky.logger.warning("Could not open update log: %s", exc) self._log_file = None def _close_update_log(self): if self._log_file: try: self._log_file.close() except OSError: pass self._log_file = None async def _log(self, line): line = line.rstrip() self.state.log.append(line) if self._log_file: try: self._log_file.write(line + "\n") self._log_file.flush() except OSError: self._log_file = None await decky.emit("cachyos_update_log", line) async def _set_phase(self, phase, done_weight, total_weight): self.state.phase = phase self.state.progress = done_weight / total_weight if total_weight else 0.0 # Each phase reports its own download size. self.state.download_mib = 0.0 self.state.downloading = False await self._emit_progress() async def _emit_progress(self): # One object rather than a growing list of positional arguments. await decky.emit( "cachyos_update_progress", { "progress": self.state.progress, "phase": self.state.phase, "update_started": self.state.update_started, "download_mib": self.state.download_mib, "downloading": self.state.downloading, }, ) async def _set_status(self, status): self.state.status = status await decky.emit("cachyos_update_state", status) # -- temporary pacman permission --------------------------------------- async def _grant_pacman(self, user): """Let the desktop user call pacman without a password, briefly. Building AUR packages has to happen unprivileged, but the install step pacman performs needs root. Running yay as root instead is not an option: it builds inside a systemd DynamicUser sandbox whose directory lives under /var/cache/private, which is unreachable for the build itself, so anything that compiles fails. The rule is therefore written just before the AUR phase and removed immediately after, in a finally block, on plugin start and on unload. It is scoped to pacman, but note that this still amounts to root for the duration - pacman can install arbitrary packages, and package scripts run as root. """ staging = Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "sudoers-staging" try: staging.write_text(f"{user} ALL=(root) NOPASSWD: /usr/bin/pacman\n") staging.chmod(0o440) except OSError as exc: decky.logger.error("Could not stage sudoers rule: %s", exc) return False # A malformed file in sudoers.d breaks sudo for everything, so never # install one that visudo has not accepted. rc, lines = await self._run( ["visudo", "-c", "-f", str(staging)], stream=False ) if rc != 0: decky.logger.error("Refusing invalid sudoers rule: %s", " ".join(lines)) staging.unlink(missing_ok=True) return False try: shutil.copyfile(staging, SUDOERS_FILE) os.chown(SUDOERS_FILE, 0, 0) os.chmod(SUDOERS_FILE, 0o440) except OSError as exc: decky.logger.error("Could not install sudoers rule: %s", exc) self._revoke_pacman() return False finally: staging.unlink(missing_ok=True) return True def _revoke_pacman(self): try: os.remove(SUDOERS_FILE) decky.logger.info("Temporary pacman permission removed") except FileNotFoundError: pass except OSError as exc: decky.logger.error( "COULD NOT REMOVE %s - please delete it manually: %s", SUDOERS_FILE, exc, ) # -- pacman lock ------------------------------------------------------- async def _heal_pacman_lock(self): """Clear a leftover lock, so an interrupted run cannot block later ones. Called on plugin start, before and after every update and before every check: whichever of those happens first repairs the system, without the user ever having to know that /var/lib/pacman/db.lck exists. """ if self._dry_run: return False cleared = _clear_stale_lock() if not cleared: return False await self._log( f"!! Removed stale pacman lock {PACMAN_LOCK} " f"({int(cleared['age'])} s old, no process had it open)" ) return True # -- process runner ---------------------------------------------------- async def _run( self, cmd, env=None, stream=True, phase_base=0.0, phase_span=0.0, merge_stderr=True, ): """Run a command, streaming stdout(+stderr) line by line. Returns (returncode, list_of_lines). `merge_stderr=False` is for probes whose output gets parsed - fwupdmgr in particular interleaves progress ticks on stderr that would otherwise corrupt the result. """ if self._dry_run: await self._log("$ " + " ".join(cmd)) return 0, [] if stream: await self._log("$ " + " ".join(cmd)) try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=( asyncio.subprocess.STDOUT if merge_stderr else asyncio.subprocess.DEVNULL ), stdin=asyncio.subprocess.DEVNULL, env=env or _base_env(), ) except FileNotFoundError: if stream: await self._log(f"!! {cmd[0]} not found") return 127, [] lines = [] assert proc.stdout is not None while True: raw = await proc.stdout.readline() if not raw: break line = ANSI_RE.sub("", raw.decode("utf-8", "replace").rstrip("\n")) # Carriage returns redraw a line in place; keep only what would # actually be visible. if "\r" in line: line = line.rsplit("\r", 1)[-1] lines.append(line) if stream: await self._log(line) size = DOWNLOAD_SIZE_RE.search(line) if size: self.state.download_mib = float(size.group(1)) await self._emit_progress() elif line.startswith(":: Retrieving packages"): # Nothing is printed again until the first package is # installed, so let the bar animate instead of freezing. self.state.downloading = True await self._emit_progress() # pacman/yay print "(12/34) upgrading foo" - use it to move the # bar inside a phase instead of only between phases. if phase_span: match = PROGRESS_RE.search(line) if match: cur, total = int(match.group(1)), int(match.group(2)) if total: frac = min(cur / total, 1.0) self.state.progress = phase_base + phase_span * frac self.state.downloading = False await self._emit_progress() rc = await proc.wait() return rc, lines # -- checking ---------------------------------------------------------- async def _check_pacman(self): if not _which("checkupdates"): return [], "missing_checkupdates" # A stable database path under the desktop user's cache, so the sync # is not repeated on every call. It has to be writable by that user, # since checkupdates runs unprivileged. _, home = _desktop_user() cache = f"{home}/.cache/decky-cachyos-update" if home else "/tmp" cmd, env = _user_cmd( ["checkupdates"], {"CHECKUPDATES_DB": f"{cache}/checkupdates-db"} ) rc, lines = await self._run(cmd, env=env, stream=False) # checkupdates is inverted: 0 = updates found, 2 = nothing to do. if rc == 2: return [], "" if rc != 0: # Surface the reason instead of only a generic warning - this is # the one check that has several unrelated ways to fail. for line in lines[-3:]: if line.strip(): await self._log(f"!! checkupdates: {line.strip()}") return [], "checkupdates_failed" return [l for l in lines if l.strip()], "" async def _check_aur(self): if not self.settings["enable_aur"]: return [], "" yay = _which("yay") if not yay: return [], "missing_yay" cmd, env = _user_cmd([yay, "-Qua", "--color", "never"]) rc, lines = await self._run(cmd, env=env, stream=False) entries = [l for l in lines if AUR_UPDATE_RE.match(l)] if rc != 0 and not entries: return [], "" return entries, "" async def _check_flatpak(self): if not self.settings["enable_flatpak"]: return [], "" flatpak = _which("flatpak") if not flatpak: return [], "" entries = [] # Refreshing appstream data is what makes --updates accurate. await self._run( [flatpak, "update", "--system", "--appstream", "--noninteractive"], stream=False, ) rc, lines = await self._run( [ flatpak, "remote-ls", "--system", "--updates", "--columns=application,version", ], stream=False, ) if rc == 0: entries += [l for l in lines if l.strip()] if _can_run_as_user(): cmd, env = _user_cmd( [flatpak, "remote-ls", "--user", "--updates", "--columns=application,version"] ) rc, lines = await self._run(cmd, env=env, stream=False) if rc == 0: entries += [l for l in lines if l.strip()] return entries, "" async def _check_fwupd(self): if not self.settings["enable_fwupd"]: return [], "" fwupdmgr = _which("fwupdmgr") if not fwupdmgr: return [], "" rc, lines = await self._run( [fwupdmgr, "get-updates", "--json"], stream=False, merge_stderr=False ) # With --json, "nothing to do" is exit 0 plus an empty Devices array; # older fwupd returns 2 (EXIT_NOTHING_TO_DO). Anything else is unusable. if rc not in (0, 2): return [], "" try: data = json.loads("\n".join(lines)) except ValueError: return [], "" entries = [] for dev in data.get("Devices", []): name = dev.get("Name", "?") for rel in dev.get("Releases", []): entries.append(f"{name} -> {rel.get('Version', '?')}") return entries, "" async def check_updates(self): """Look for updates without installing anything.""" if self.lock.locked(): return self.state.to_dict() async with self.lock: await self._set_status("checking") # The periodic check is what runs most often, so letting it repair # the lock means an abandoned one is usually gone long before the # user ever presses "Update". await self._heal_pacman_lock() warnings = [] try: results = {} for key, coro in ( ("pacman", self._check_pacman()), ("aur", self._check_aur()), ("flatpak", self._check_flatpak()), ("fwupd", self._check_fwupd()), ): entries, warning = await coro results[key] = entries if warning: warnings.append(warning) self.state.details = results self.state.counts = {k: len(v) for k, v in results.items()} self.state.last_check = time.time() self.state.warnings = warnings self.state.error_text = "" self._save_state() except Exception as exc: decky.logger.exception("Update check failed") self.state.error_text = str(exc) finally: await self._set_status("idle") return self.state.to_dict() # -- updating ---------------------------------------------------------- def _enabled_phases(self): phases = ["keyring", "pacman"] if self.settings["enable_aur"] and _which("yay"): phases.append("aur") if self.settings["enable_flatpak"] and _which("flatpak"): phases.append("flatpak_system") if _can_run_as_user(): phases.append("flatpak_user") if self.settings["enable_fwupd"] and _which("fwupdmgr"): phases.append("fwupd") return phases async def _repo_version(self, pkg): rc, lines = await self._run(["pacman", "-Si", pkg], stream=False) if rc != 0: return "" for line in lines: if line.startswith("Version"): # "Version : 1:20260707.1-1" - split once, epochs contain ":". return line.split(":", 1)[1].strip() return "" async def _local_version(self, pkg): rc, lines = await self._run(["pacman", "-Q", pkg], stream=False) parts = lines[0].split() if rc == 0 and lines else [] return parts[1] if len(parts) > 1 else "" async def _keyrings_to_upgrade(self): """Keyring packages the repositories actually offer as an upgrade. Guards against two situations seen in the wild: the package missing from the repositories entirely (pacman would abort the whole transaction with "target not found"), and the repository carrying an older version than what is installed, which --needed does not catch and which would silently downgrade the keyring. """ upgradable = [] for pkg in ("archlinux-keyring", "cachyos-keyring"): if not _installed(pkg): continue repo = await self._repo_version(pkg) local = await self._local_version(pkg) if not repo or not local: continue rc, lines = await self._run(["vercmp", local, repo], stream=False) try: if rc == 0 and int(lines[0].strip()) < 0: upgradable.append(pkg) except (IndexError, ValueError): continue return upgradable async def _phase_keyring(self, **kw): # Refreshing the databases is the part that has to work. rc, lines = await self._run( ["pacman", "-Sy", "--noconfirm", "--color", "never"], **kw ) if rc != 0: return False, lines # Upgrading the keyring first is the documented cure for "invalid or # corrupted package (PGP signature)". It is only a precaution, so it # must never stop the actual upgrade from happening. packages = await self._keyrings_to_upgrade() if packages: rc, extra = await self._run( ["pacman", "-S", "--needed", "--noconfirm", "--color", "never"] + packages, **kw, ) lines += extra if rc != 0: await self._log("!! Keyring update failed, continuing anyway.") return True, lines async def _phase_pacman(self, **kw): cmd = [ "pacman", "-Su", "--noconfirm", "--noprogressbar", "--color", "never", ] rc, output = await self._run(cmd, **kw) # One package the repositories cannot agree on must not cost the user # all the other updates. Waiting for the missing rebuild is the only # real cure, so hold that package back and upgrade everything else in # the meantime. Each retry is a complete transaction that pacman # validates on its own, so the system stays consistent - it is simply # one package behind until the rebuild lands. # Only the blocker is kept out of pacman's way; its dependant has no # repository upgrade to skip. Both go into state.held_packages, which # is what the user is shown and what yay is told to leave alone. blockers = [] history = list(output) while rc != 0 and len(blockers) < MAX_HELD_PACKAGES: fresh = [ (b, d) for b, d in _dependency_breaks(history) if b not in blockers ] if not fresh: break for blocker, dependant in fresh: if blocker not in blockers: blockers.append(blocker) for name in (blocker, dependant): if name not in self.state.held_packages: self.state.held_packages.append(name) await self._log( "!! Holding back " + ", ".join(f"{b} (pinned by {d})" for b, d in fresh) + " - upgrading everything else." ) rc, output = await self._run( cmd + ["--ignore", ",".join(blockers)], **kw ) history = history + output if rc != 0: self.state.held_packages = [] return False, history # The attempts that were abandoned installed nothing, and their error # lines would otherwise make a later phase look like it broke over a # dependency. Only what actually ran is worth reporting. return True, output async def _phase_aur(self, **kw): args = [ # --aur, not a second full -Syu: the repository upgrade already # happened in the pacman phase. Repeating it here re-resolves the # dependency break that phase worked around, and this time the # blocker arrives as an explicit target - for which pacman only # asks "in IgnorePkg, install anyway?" and --noconfirm says yes. _which("yay") or "yay", "-Syu", "--aur", "--noconfirm", "--removemake", "--noprogressbar", "--color", "never", # Fail fast instead of blocking on a password prompt that no one # can answer if the sudoers rule did not take effect. "--sudoflags", "-n", ] # A held-back pair stays held back here too. Without this yay would # rebuild the dependant, pull the blocker in as its dependency, and # hit the very conflict the pacman phase stepped around. if self.state.held_packages: args += ["--ignore", ",".join(self.state.held_packages)] cmd, env = _user_cmd(args) user, _ = _desktop_user() granted = False if _is_root() and user: granted = await self._grant_pacman(user) if not granted: await self._log("!! Could not grant pacman access - skipping AUR.") return False, [] try: rc, lines = await self._run(cmd, env=env, **kw) finally: if granted: self._revoke_pacman() return rc == 0, lines async def _phase_flatpak_system(self, **kw): # --system is not optional here: without an explicit scope flatpak # updates the system installation *and the calling user's* one, which # as root means it would start populating /root/.local/share/flatpak. # --noninteractive already implies -y and picks the quiet transaction. rc, lines = await self._run( [ _which("flatpak") or "flatpak", "update", "--system", "--noninteractive", ], **kw, ) return rc == 0, lines async def _phase_flatpak_user(self, **kw): cmd, env = _user_cmd( [ _which("flatpak") or "flatpak", "update", "--user", "--noninteractive", ] ) rc, lines = await self._run(cmd, env=env, **kw) return rc == 0, lines async def _phase_fwupd(self, **kw): fwupdmgr = _which("fwupdmgr") or "fwupdmgr" # No --force: it would re-download the metadata on every single run, # which is wasteful on a battery-powered handheld. Fresh metadata # simply exits 2 (EXIT_NOTHING_TO_DO), which is a success here. rc, lines = await self._run([fwupdmgr, "refresh", "--assume-yes"], **kw) if rc == 101: # ENETUNREACH await self._log("!! Firmware server unreachable - skipped.") return True, lines rc, update_lines = await self._run( [ fwupdmgr, "update", "--assume-yes", "--no-reboot-check", "--no-unreported-check", "--no-remote-check", ], **kw, ) lines += update_lines # 0 = done, 2 = EXIT_NOTHING_TO_DO, 101 = network unreachable. return rc in (0, 2, 101), lines async def start_update(self, dry_run=False): if self.lock.locked(): return {"started": False, "reason_id": "already_running", "holder": ""} if not _is_root() and not dry_run: return {"started": False, "reason_id": "not_root", "holder": ""} if not dry_run: # An abandoned lock must never be a reason to refuse: clear it # first, and only report "busy" for one that is genuinely held. _clear_stale_lock() status = _lock_status() if status["held"]: return { "started": False, "reason_id": "pacman_busy", "holder": status["holder"], } asyncio.create_task(self._do_update(dry_run)) return {"started": True, "reason_id": "", "holder": ""} async def _do_update(self, dry_run=False): async with self.lock: self._dry_run = dry_run self.state.log.clear() self.state.error_text = "" self.state.hint_id = "" self.state.warnings = [] self.state.failed_phases = [] self.state.failed_packages = [] self.state.held_packages = [] self.state.progress = 0.0 self.state.download_mib = 0.0 self.state.downloading = False self.state.update_started = time.time() self._open_update_log() await self._set_status("updating") await self._heal_pacman_lock() phases = self._enabled_phases() total_weight = sum(PHASE_WEIGHTS[p] for p in phases) done_weight = 0 all_output = [] # Kept apart from all_output: a phase that recovered on its own # still printed the error it recovered from, and that error must # not end up explaining a later phase's failure. failed_output = [] handlers = { "keyring": self._phase_keyring, "pacman": self._phase_pacman, "aur": self._phase_aur, "flatpak_system": self._phase_flatpak_system, "flatpak_user": self._phase_flatpak_user, "fwupd": self._phase_fwupd, } try: for phase in phases: weight = PHASE_WEIGHTS[phase] await self._set_phase(phase, done_weight, total_weight) ok, lines = await handlers[phase]( phase_base=done_weight / total_weight, phase_span=weight / total_weight, ) all_output += lines done_weight += weight if not ok: self.state.failed_phases.append(phase) failed_output += lines await self._log(f"!! Phase failed: {phase}") # The repo upgrade is the foundation - if it breaks, # building AUR packages on top of it is a bad idea. if phase in ("keyring", "pacman"): break self.state.progress = 1.0 _invalidate_installed_cache() self.state.reboot_required = self._needs_reboot(all_output) if "fwupd" in phases and not self.state.reboot_required: self.state.reboot_required = await self._firmware_reboot_pending() self.state.last_update = time.time() if self.state.failed_phases: self.state.status = "error" self.state.hint_id = self._hint_for(failed_output) self.state.failed_packages = self._failed_packages(failed_output) else: self.state.status = "done" except Exception as exc: decky.logger.exception("Update failed") self.state.status = "error" self.state.error_text = str(exc) finally: # A pacman killed mid-run - suspend, OOM, a forced reload of # the plugin - leaves its lock behind. Clean up after # ourselves instead of handing the problem to the next run. await self._heal_pacman_lock() self._dry_run = False self._close_update_log() await decky.emit("cachyos_update_state", self.state.status) await decky.emit( "cachyos_update_finished", self.state.status == "done", self.state.failed_phases, self.state.reboot_required, ) # Only a successful run means "nothing pending" - after a failure # the old counts stay, otherwise the panel would claim the system # is up to date while the updates are still sitting there. if not dry_run and self.state.status == "done": self.state.details = {k: [] for k in self.state.details} self.state.counts = {k: 0 for k in self.state.counts} self._save_state() # -- post-update analysis --------------------------------------------- def _failed_packages(self, lines): """Names yay reported as unbuildable, in order, without duplicates.""" found = [] collecting = False for line in lines: if FAILED_HEADER in line: collecting = True continue if not collecting: continue match = FAILED_PKG_RE.match(line.strip()) if match: name = match.group(1) if name not in found: found.append(name) elif line.strip(): collecting = False return found def _hint_for(self, lines): blob = "\n".join(lines).lower() for marker, hint_id in ERROR_HINTS: if marker in blob: return hint_id return "" def _needs_reboot(self, output_lines): """True if the running kernel no longer matches what is installed. Primary signal is the same one cachy-update uses: the module tree of the running kernel still has its vmlinuz. Skipped when kernel-modules-hook or mkmm is installed, because those keep old module trees around and would make the test always pass. """ try: if not (_installed("kernel-modules-hook") or _installed("mkmm")): release = os.uname().release if not Path(f"/usr/lib/modules/{release}/vmlinuz").is_file(): return True except Exception as exc: decky.logger.warning("Kernel check failed: %s", exc) # Cross-check: did this run upgrade anything from the reboot list? upgraded = { match.group(1) for line in output_lines for match in [UPGRADING_RE.search(line)] if match } return any( pattern.match(pkg) for pkg in upgraded for pattern in REBOOT_TRIGGER_PATTERNS ) async def _firmware_reboot_pending(self): """Firmware is staged to the ESP and flashed on the next boot. `--no-reboot-check` suppresses the prompt but not the need, so ask fwupd explicitly. Exit 0 means a reboot IS pending, 2 means clean. """ fwupdmgr = _which("fwupdmgr") if not fwupdmgr or self._dry_run: return False rc, _ = await self._run( [fwupdmgr, "check-reboot-needed"], stream=False, merge_stderr=False ) return rc == 0 # -- frontend API ------------------------------------------------------ async def get_state(self): return self.state.to_dict() async def get_settings(self): return self.settings async def set_settings(self, settings): for key in DEFAULT_SETTINGS: if key in settings: self.settings[key] = settings[key] self._save_settings() return self.settings async def reboot(self): if self._dry_run: return True await self._run(["systemctl", "reboot"], stream=False) return True async def self_test(self): tools = {} # (binary, version args, substrings that identify the relevant line) for name, args, match in ( ("pacman", ["-Q", "pacman"], ()), ("yay", ["--version"], ()), ("paru", ["--version"], ()), ("checkupdates", None, ()), ("flatpak", ["--version"], ()), # fwupdmgr prints a compile/runtime table, not a single line. ("fwupdmgr", ["--version"], ("runtime", "org.freedesktop.fwupd")), ("runuser", None, ()), ("sudo", None, ()), ("visudo", None, ()), ("systemd-run", ["--version"], ()), ("fakeroot", None, ()), ): path = _which(name) entry = {"found": bool(path), "path": path or "", "version": ""} if path and args: try: rc, lines = await self._run( [path] + args, stream=False, merge_stderr=False ) if rc == 0: for line in lines: text = " ".join(line.split()) if not text: continue if match: if all(m in text for m in match): entry["version"] = text[:80] break elif any(c.isdigit() for c in text): entry["version"] = text[:80] break except Exception: pass tools[name] = entry lock = _lock_status() return { "root": _is_root(), "uid": os.geteuid(), "standalone": STANDALONE, "decky_user": getattr(decky, "DECKY_USER", ""), "decky_user_home": getattr(decky, "DECKY_USER_HOME", ""), "kernel": os.uname().release, "kernel_vmlinuz_present": Path( f"/usr/lib/modules/{os.uname().release}/vmlinuz" ).is_file(), "pacman_lock": { "present": lock["present"], "held": lock["held"], "stale": lock["stale"], "holder": lock["holder"], "holder_pid": lock["holder_pid"], "age": int(lock["age"]), }, "tools": tools, "settings": self.settings, } # -- background check -------------------------------------------------- async def _auto_check_loop(self): # Let the session settle before hitting the network. await asyncio.sleep(30) while True: try: if self.settings["auto_check"]: await self.check_updates() total = sum(self.state.counts.values()) threshold = max(1, int(self.settings.get("notify_threshold", 1))) # Announce unconditionally and let the frontend decide what # to do with it: only that side knows whether a game is # running, and automatic installs must not be tied to # whether the user wants toasts. if total: await decky.emit( "cachyos_update_available", total, total >= threshold and self.settings["notify_on_updates"], ) except asyncio.CancelledError: raise except Exception as exc: decky.logger.warning("Automatic check failed: %s", exc) hours = max(1, int(self.settings.get("check_interval_hours", 6))) await asyncio.sleep(hours * 3600) # -------------------------------------------------------------------------- # standalone entry point (development only) # -------------------------------------------------------------------------- async def _cli(): plugin = Plugin() plugin._load_settings() args = sys.argv[1:] if "--selftest" in args: result = await plugin.self_test() print(json.dumps(result, indent=2, ensure_ascii=False)) if not result["root"]: print("\nNote: not running as root - expected during development.") return if "--fix-lock" in args: status = _lock_status() if not status["present"]: print(f"{PACMAN_LOCK}: not present - pacman is free to run.") elif status["held"]: print( f"{PACMAN_LOCK}: held by {status['holder'] or 'another process'}" f" (pid {status['holder_pid'] or '?'}), {int(status['age'])} s old." "\nNothing to do - wait for it to finish." ) elif _clear_stale_lock(): print(f"{PACMAN_LOCK}: stale lock removed.") else: print(f"{PACMAN_LOCK}: present, {int(status['age'])} s old - left alone.") return if "--check" in args: state = await plugin.check_updates() print(json.dumps( { "counts": state["counts"], "total": state["total"], "warnings": state["warnings"], "details": state["details"], }, indent=2, ensure_ascii=False, )) return if "--update" in args: dry = "--dry-run" in args if not dry and not _is_root(): print("A real update needs root. Use --dry-run to test.") return if dry: # Enable every phase so the dry run shows all command lines. plugin.settings["enable_fwupd"] = True await plugin._do_update(dry_run=dry) print(f"\nStatus: {plugin.state.status}") if plugin.state.held_packages: print(f"Held back: {', '.join(plugin.state.held_packages)}") if plugin.state.failed_phases: print(f"Failed phases: {', '.join(plugin.state.failed_phases)}") if plugin.state.error_text: print(f"Error: {plugin.state.error_text}") return print(__doc__) if __name__ == "__main__": asyncio.run(_cli())