Gabriel Brown
4cbab01ae2
Automatic updates leave no other trace. The Flatpak that sat here as "1 update available" installed itself at 00:14 this morning and nothing on the machine would have said so. Both sources are asked in their own machine-readable form and merged on time, so the answer reads as one history rather than two lists to interleave by eye. Two parsing traps worth recording next to the code. flatpak's --json prints timestamps as "Aug 20 08:07:46" with no year in them, so the year is inferred and a date that would land in the future is read as last year's. And dnf5's start_time is epoch UTC while its own history table prints that same value as though it were local -- checked against rpm, and the local rendering here is the correct one. The contract asserts entries are newest first, that none is dated in the future, and that both sources parse; it was verified to fail by breaking the year inference so every flatpak entry landed tomorrow. Loaded on demand rather than with the page, because it reads both full transaction logs. Claude-Session: https://claude.ai/code/session_01BRvzt4H8XXLPVH5MyYdk9L
415 lines
16 KiB
Python
Executable File
415 lines
16 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 datetime
|
|
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 flatpak_time(text: str) -> int:
|
|
"""flatpak prints "Aug 20 08:07:46" -- a time with no year in it.
|
|
|
|
Assumed to be this year, and rolled back one if that would put it in the
|
|
future, which is the only reading that makes sense for a history. Anything
|
|
unparseable sorts last rather than pretending to be the epoch, which would
|
|
put it at the top of a newest-first list.
|
|
"""
|
|
if not text.strip():
|
|
return 0
|
|
now = datetime.datetime.now()
|
|
for year in (now.year, now.year - 1):
|
|
try:
|
|
when = datetime.datetime.strptime(f"{year} {text.strip()}", "%Y %b %d %H:%M:%S")
|
|
except ValueError:
|
|
return 0
|
|
if when <= now + datetime.timedelta(days=1):
|
|
return int(when.timestamp())
|
|
return 0
|
|
|
|
|
|
def history(limit: int = 25) -> list[dict]:
|
|
"""What has actually been installed, newest first.
|
|
|
|
Both sources are asked in their own machine-readable form rather than by
|
|
parsing their tables: dnf5 prints JSON, and flatpak takes --json. The two
|
|
are merged on time so the answer reads as one history rather than as two
|
|
lists a person has to interleave themselves.
|
|
|
|
Automatic updates are the reason this is worth showing. Something that
|
|
installs itself overnight leaves no other trace a person would notice.
|
|
"""
|
|
entries: list[dict] = []
|
|
|
|
dnf = run(["dnf5", "history", "list", "--json"], timeout=30.0)
|
|
if dnf.returncode == 0:
|
|
try:
|
|
for row in json.loads(dnf.stdout or "[]"):
|
|
command = str(row.get("command_line") or "").strip()
|
|
entries.append({
|
|
"source": "dnf",
|
|
"at": int(row.get("start_time") or 0),
|
|
# The full argv is noise; what was done is the useful part.
|
|
"summary": command.split("/")[-1] if command else "transaction",
|
|
"count": int(row.get("altered_count") or 0),
|
|
"ok": str(row.get("status") or "") == "Ok",
|
|
})
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
pass
|
|
|
|
flatpak = run(["flatpak", "history", "--json"], timeout=30.0)
|
|
if flatpak.returncode == 0:
|
|
try:
|
|
rows = json.loads(flatpak.stdout or "[]")
|
|
except json.JSONDecodeError:
|
|
rows = []
|
|
for row in rows if isinstance(rows, list) else []:
|
|
application = str(row.get("application") or "").strip()
|
|
change = str(row.get("change") or "").strip()
|
|
if not application:
|
|
continue
|
|
# "deploy update", "deploy install", "uninstall" -- the verb is a
|
|
# word inside the change rather than the whole of it.
|
|
verb = next((word for word in ("update", "install", "uninstall")
|
|
if word in change), "")
|
|
if not verb:
|
|
continue
|
|
entries.append({
|
|
"source": "flatpak",
|
|
"at": flatpak_time(str(row.get("time") or "")),
|
|
"summary": verb + " " + application,
|
|
"count": 1,
|
|
"ok": True,
|
|
})
|
|
|
|
entries.sort(key=lambda entry: entry["at"], reverse=True)
|
|
return entries[:limit]
|
|
|
|
|
|
def main(arguments: list[str]) -> int:
|
|
try:
|
|
if arguments == ["history"]:
|
|
print(json.dumps({"entries": history(), "error": ""}, separators=(",", ":")))
|
|
return 0
|
|
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:]))
|