Gabriel Brown
a15f019c17
The page had a row for the case where dnf-automatic is absent and nothing for the case where it is present, so installing it made the setting disappear rather than appear. The switch enables downloading only, which is the shipped default and the right one to keep: a machine that installs packages unattended can reboot into a kernel nobody chose. Downloading ahead of time is what makes the install quick when someone does choose it, and the row says exactly that rather than implying updates apply themselves. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
332 lines
13 KiB
Python
Executable File
332 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Software updates, from every source this machine actually uses.
|
|
|
|
Three of them, and they fail independently, so they are counted and applied
|
|
separately rather than blended into one number: packages (dnf), applications
|
|
(flatpak), and firmware (fwupd).
|
|
|
|
Checking costs about nine seconds of network and metadata work, which is too
|
|
long to spend every time a page opens. So `snapshot` is instant -- it reads the
|
|
last result plus the things that are free to compute -- and `check` is the scan
|
|
that refreshes it. The page shows when it last checked, the way every mature
|
|
updater does, instead of pretending the number is live.
|
|
|
|
panama-updates snapshot
|
|
panama-updates check
|
|
panama-updates apply dnf|flatpak|firmware
|
|
panama-updates set-auto-flatpak true|false
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# The user timer this ships for keeping applications current. dnf has no
|
|
# equivalent here because dnf-automatic is not installed, and installing
|
|
# software is not this script's job.
|
|
FLATPAK_TIMER = "panama-flatpak-update.timer"
|
|
|
|
# Anything carrying an advisory of these severities is reported as a security
|
|
# fix. "none" is excluded deliberately: an advisory with no severity is a
|
|
# bugfix or enhancement, and calling it security would cry wolf.
|
|
SECURITY_SEVERITIES = "critical,important,moderate,low"
|
|
|
|
|
|
class BoundaryError(RuntimeError):
|
|
"""A user-visible failure."""
|
|
|
|
|
|
def run(command: list[str], timeout: float = 180.0) -> subprocess.CompletedProcess:
|
|
try:
|
|
return subprocess.run(command, capture_output=True, text=True,
|
|
timeout=timeout, check=False)
|
|
except subprocess.TimeoutExpired as error:
|
|
raise BoundaryError(f"{command[0]} did not finish in time.") from error
|
|
except OSError as error:
|
|
raise BoundaryError(f"{command[0]} is not available.") from error
|
|
|
|
|
|
def cache_path() -> Path:
|
|
base = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "panama"
|
|
base.mkdir(parents=True, exist_ok=True)
|
|
return base / "updates.json"
|
|
|
|
|
|
def read_cache() -> dict:
|
|
try:
|
|
return json.loads(cache_path().read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
|
|
|
|
def write_cache(payload: dict) -> None:
|
|
# Written atomically: a page reading this while it is half-written would
|
|
# report zero updates, which is the one wrong answer that looks fine.
|
|
target = cache_path()
|
|
temporary = target.with_suffix(".tmp")
|
|
try:
|
|
temporary.write_text(json.dumps(payload), encoding="utf-8")
|
|
temporary.replace(target)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def kernel_state() -> dict:
|
|
"""Whether a reboot would change the kernel you are running.
|
|
|
|
This is the honest version of "restart required". Comparing the running
|
|
release against the newest installed one is exact, needs no plugin, and
|
|
takes no time -- and a machine that has already rebooted since the update
|
|
correctly reports nothing pending.
|
|
"""
|
|
running = os.uname().release
|
|
newest = running
|
|
result = run(["rpm", "-q", "kernel", "--qf", "%{VERSION}-%{RELEASE}.%{ARCH}\\n"], timeout=30)
|
|
if result.returncode == 0:
|
|
installed = [line.strip() for line in result.stdout.splitlines() if line.strip()]
|
|
if installed:
|
|
# rpm lists oldest first for equal names.
|
|
newest = installed[-1]
|
|
return {
|
|
"running": running,
|
|
"newestInstalled": newest,
|
|
"rebootNeeded": newest != running,
|
|
}
|
|
|
|
|
|
def dnf_updates() -> dict:
|
|
if not shutil.which("dnf5"):
|
|
return {"available": False, "count": 0, "packages": [], "securityCount": 0}
|
|
|
|
result = run(["dnf5", "check-upgrade", "--json"], timeout=180)
|
|
packages = []
|
|
# dnf5 exits 100 when upgrades exist, 0 when none do. Both are success.
|
|
if result.returncode in (0, 100):
|
|
try:
|
|
payload = json.loads(result.stdout or "{}")
|
|
for entry in payload.get("upgrades", []):
|
|
packages.append({
|
|
"name": str(entry.get("name", "")),
|
|
"version": str(entry.get("evr", "")),
|
|
"repository": str(entry.get("repository", "")),
|
|
})
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
security = 0
|
|
advisory = run(["dnf5", "check-upgrade",
|
|
f"--advisory-severities={SECURITY_SEVERITIES}", "--json"], timeout=180)
|
|
if advisory.returncode in (0, 100):
|
|
try:
|
|
security = len(json.loads(advisory.stdout or "{}").get("upgrades", []))
|
|
except json.JSONDecodeError:
|
|
security = 0
|
|
|
|
packages.sort(key=lambda item: item["name"])
|
|
return {"available": True, "count": len(packages), "packages": packages,
|
|
"securityCount": security}
|
|
|
|
|
|
def flatpak_updates() -> dict:
|
|
if not shutil.which("flatpak"):
|
|
return {"available": False, "count": 0, "applications": []}
|
|
result = run(["flatpak", "remote-ls", "--updates", "--columns=application,version"],
|
|
timeout=120)
|
|
applications = []
|
|
if result.returncode == 0:
|
|
for line in result.stdout.splitlines():
|
|
parts = [part.strip() for part in line.split("\t")]
|
|
if parts and parts[0]:
|
|
applications.append({"id": parts[0],
|
|
"version": parts[1] if len(parts) > 1 else ""})
|
|
return {"available": True, "count": len(applications), "applications": applications}
|
|
|
|
|
|
def firmware_updates() -> dict:
|
|
if not shutil.which("fwupdmgr"):
|
|
return {"available": False, "count": 0, "devices": []}
|
|
result = run(["fwupdmgr", "get-updates", "--json"], timeout=120)
|
|
devices = []
|
|
try:
|
|
payload = json.loads(result.stdout or "{}")
|
|
for device in payload.get("Devices", []):
|
|
releases = device.get("Releases", [])
|
|
devices.append({
|
|
"name": str(device.get("Name", "Unknown device")),
|
|
"version": str(device.get("Version", "")),
|
|
"target": str(releases[0].get("Version", "")) if releases else "",
|
|
# Firmware that needs a reboot to flash is worth saying up front.
|
|
"needsReboot": "needs-reboot" in json.dumps(device.get("Flags", [])),
|
|
})
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return {"available": True, "count": len(devices), "devices": devices}
|
|
|
|
|
|
def automatic_state() -> dict:
|
|
flatpak_timer = run(["systemctl", "--user", "is-enabled", FLATPAK_TIMER], timeout=20)
|
|
dnf_timer = run(["systemctl", "is-enabled", "dnf5-automatic.timer"], timeout=20)
|
|
return {
|
|
"flatpakEnabled": flatpak_timer.stdout.strip() == "enabled",
|
|
"flatpakAvailable": flatpak_timer.stdout.strip() not in ("", "not-found"),
|
|
# Reported, never offered: dnf-automatic is a package this machine does
|
|
# not have, and installing software is not a settings action.
|
|
"dnfAutomaticEnabled": dnf_timer.stdout.strip() == "enabled",
|
|
"dnfAutomaticAvailable": dnf_timer.stdout.strip() not in ("", "not-found"),
|
|
}
|
|
|
|
|
|
def check() -> dict:
|
|
payload = {
|
|
"dnf": dnf_updates(),
|
|
"flatpak": flatpak_updates(),
|
|
"firmware": firmware_updates(),
|
|
"checkedAt": int(time.time()),
|
|
}
|
|
write_cache(payload)
|
|
return payload
|
|
|
|
|
|
def snapshot() -> dict:
|
|
cached = read_cache()
|
|
empty = {"available": True, "count": 0}
|
|
return {
|
|
"dnf": cached.get("dnf", {**empty, "packages": [], "securityCount": 0}),
|
|
"flatpak": cached.get("flatpak", {**empty, "applications": []}),
|
|
"firmware": cached.get("firmware", {**empty, "devices": []}),
|
|
# 0 means never checked, which the page says rather than showing a
|
|
# confident "0 updates" it has no basis for.
|
|
"checkedAt": int(cached.get("checkedAt", 0)),
|
|
"kernel": kernel_state(),
|
|
"automatic": automatic_state(),
|
|
"error": "",
|
|
}
|
|
|
|
|
|
def take_restore_point(reason: str) -> str:
|
|
"""A snapshot before the system changes, named after what is about to happen.
|
|
|
|
Best effort: if snapper is not configured, the update still proceeds. An
|
|
update that refuses to run because a nicety failed would be worse than one
|
|
without a restore point.
|
|
"""
|
|
if not shutil.which("snapper"):
|
|
return ""
|
|
result = run(["snapper", "-c", "root", "create", "--description", reason,
|
|
"--cleanup-algorithm", "number", "--print-number"], timeout=120)
|
|
return result.stdout.strip() if result.returncode == 0 else ""
|
|
|
|
|
|
def apply(source: str) -> dict:
|
|
if source == "flatpak":
|
|
if not shutil.which("flatpak"):
|
|
raise BoundaryError("Flatpak is not installed.")
|
|
result = run(["flatpak", "update", "-y", "--noninteractive"], timeout=3600)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "The applications could not be updated."))
|
|
return {"restorePoint": ""}
|
|
|
|
if source == "dnf":
|
|
if not shutil.which("dnf5"):
|
|
raise BoundaryError("dnf is not installed.")
|
|
pending = read_cache().get("dnf", {}).get("count", 0)
|
|
restore_point = take_restore_point(
|
|
f"before {pending} package update{'' if pending == 1 else 's'}")
|
|
result = run(["pkexec", "dnf5", "upgrade", "-y"], timeout=7200)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "The packages could not be updated."))
|
|
return {"restorePoint": restore_point}
|
|
|
|
if source == "firmware":
|
|
if not shutil.which("fwupdmgr"):
|
|
raise BoundaryError("Firmware updating is not available.")
|
|
result = run(["fwupdmgr", "update", "-y", "--no-reboot-check"], timeout=3600)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "The firmware could not be updated."))
|
|
return {"restorePoint": ""}
|
|
|
|
raise BoundaryError("That is not an update source.")
|
|
|
|
|
|
def set_auto_dnf(enabled: bool) -> None:
|
|
"""Enable the packaging timer, which DOWNLOADS updates but does not apply them.
|
|
|
|
That is the shipped default (apply_updates = no) and it is the right one to
|
|
leave alone: a machine that installs packages unattended can reboot into a
|
|
kernel nobody chose. Downloading ahead of time makes the install quick when
|
|
someone does choose it.
|
|
"""
|
|
state = automatic_state()
|
|
if not state["dnfAutomaticAvailable"]:
|
|
raise BoundaryError("Automatic package updates are not installed.")
|
|
action = ["enable", "--now"] if enabled else ["disable", "--now"]
|
|
result = run(["pkexec", "systemctl", *action, "dnf5-automatic.timer"], timeout=120)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "Automatic package updates could not be changed."))
|
|
|
|
|
|
def set_auto_flatpak(enabled: bool) -> None:
|
|
action = ["enable", "--now"] if enabled else ["disable", "--now"]
|
|
result = run(["systemctl", "--user", *action, FLATPAK_TIMER], timeout=60)
|
|
if result.returncode != 0:
|
|
raise BoundaryError(_refusal(result, "Automatic application updates could not be changed."))
|
|
|
|
|
|
def _refusal(result: subprocess.CompletedProcess, fallback: str) -> str:
|
|
text = ((result.stderr or "") + "\n" + (result.stdout or "")).strip().splitlines()
|
|
meaningful = [line for line in text if line.strip()]
|
|
if not meaningful:
|
|
return fallback
|
|
last = meaningful[-1]
|
|
if "not authorized" in last.lower() or "dismissed" in last.lower():
|
|
return "That update was not authorized."
|
|
return last[:200]
|
|
|
|
|
|
def main(arguments: list[str]) -> int:
|
|
try:
|
|
if arguments == ["snapshot"]:
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
if arguments == ["check"]:
|
|
check()
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
if len(arguments) == 2 and arguments[0] == "apply":
|
|
outcome = apply(arguments[1])
|
|
# Re-check, so the page reflects what is actually left rather than
|
|
# assuming the update cleared everything it listed.
|
|
check()
|
|
state = snapshot()
|
|
state["applied"] = {"source": arguments[1], **outcome}
|
|
print(json.dumps(state, separators=(",", ":")))
|
|
return 0
|
|
if len(arguments) == 2 and arguments[0] == "set-auto-flatpak":
|
|
set_auto_flatpak(arguments[1] == "true")
|
|
elif len(arguments) == 2 and arguments[0] == "set-auto-dnf":
|
|
set_auto_dnf(arguments[1] == "true")
|
|
else:
|
|
raise BoundaryError(
|
|
"Usage: panama-updates snapshot | check | apply dnf|flatpak|firmware | "
|
|
"set-auto-flatpak true|false | set-auto-dnf true|false")
|
|
except BoundaryError as error:
|
|
state = snapshot()
|
|
state["error"] = str(error)
|
|
print(json.dumps(state, separators=(",", ":")))
|
|
return 0
|
|
|
|
print(json.dumps(snapshot(), separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|