754 lines
32 KiB
Python
Executable File
754 lines
32 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 apply flatpak <app-id>
|
|
panama-updates changelog dnf|flatpak|firmware <name>
|
|
panama-updates set-auto-flatpak true|false
|
|
panama-updates set-auto-dnf 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 two unattended-update timers this machine can have. The flatpak one is
|
|
# Panama's own user timer; the dnf one is dnf5-automatic's system timer, which
|
|
# ships with dnf5 and downloads without installing. Both are only ever enabled
|
|
# or disabled -- installing software is not this script's job, so a timer that
|
|
# is not present is reported as unavailable rather than offered.
|
|
FLATPAK_TIMER = "panama-flatpak-update.timer"
|
|
DNF_TIMER = "dnf5-automatic.timer"
|
|
|
|
# Package and application names reach a subprocess as argv, never a shell. They
|
|
# are still constrained: rpm names and flatpak application IDs both live well
|
|
# inside this, and anything outside it is not a name either source would emit.
|
|
NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]{0,127}$")
|
|
|
|
# How much changelog text is worth carrying into a settings expander. Beyond
|
|
# this it stops being something anybody reads and starts being a scroll.
|
|
CHANGELOG_LIMIT = 6000
|
|
|
|
# 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_download_sizes() -> dict[str, int]:
|
|
"""Bytes to fetch per pending package, straight from the repo metadata.
|
|
|
|
`dnf5 repoquery --queryformat` is the only place dnf5 reports this as a
|
|
number rather than as a rendered "172.9 MiB". A repo it cannot reach, or a
|
|
format it renders differently one day, yields nothing at all -- the size is
|
|
a nicety, and a wrong size is worse than no size.
|
|
"""
|
|
result = run(["dnf5", "repoquery", "--upgrades",
|
|
"--queryformat", "%{name} %{downloadsize}\\n"], timeout=180)
|
|
if result.returncode != 0:
|
|
return {}
|
|
sizes: dict[str, int] = {}
|
|
for line in result.stdout.splitlines():
|
|
parts = line.split()
|
|
if len(parts) != 2:
|
|
continue
|
|
try:
|
|
sizes[parts[0]] = int(parts[1])
|
|
except ValueError:
|
|
continue
|
|
return sizes
|
|
|
|
|
|
def dnf_updates() -> dict:
|
|
"""Pending packages, or the reason there is no count.
|
|
|
|
A failure used to be indistinguishable from an empty list: any exit code
|
|
outside 0/100 left `packages` empty and reported nothing, so a dnf that
|
|
could not reach a repository produced a confident "Up to date". Not knowing
|
|
and knowing there is nothing are different answers and are reported as
|
|
different answers.
|
|
"""
|
|
if not shutil.which("dnf5"):
|
|
return {"available": False, "count": 0, "packages": [], "securityCount": 0,
|
|
"securityKnown": True, "error": ""}
|
|
|
|
result = run(["dnf5", "check-upgrade", "--json"], timeout=180)
|
|
# dnf5 exits 100 when upgrades exist, 0 when none do. Both are success;
|
|
# anything else is dnf saying it could not answer, most often a repository
|
|
# it could not reach.
|
|
if result.returncode not in (0, 100):
|
|
return {"available": True, "count": 0, "packages": [], "securityCount": 0,
|
|
"securityKnown": False,
|
|
"error": _refusal(result, "The package list could not be read.")}
|
|
try:
|
|
payload = json.loads(result.stdout or "{}")
|
|
except json.JSONDecodeError:
|
|
return {"available": True, "count": 0, "packages": [], "securityCount": 0,
|
|
"securityKnown": False,
|
|
"error": "dnf answered with something that was not a package list."}
|
|
|
|
packages = []
|
|
for entry in payload.get("upgrades", []):
|
|
packages.append({
|
|
"name": str(entry.get("name", "")),
|
|
"version": str(entry.get("evr", "")),
|
|
"repository": str(entry.get("repository", "")),
|
|
})
|
|
|
|
# A separate question with a separate failure. This count only ever shrinks
|
|
# the alarm -- the page says "nothing security-critical" when it is zero --
|
|
# so an advisory query that failed and returned zero would be reassurance
|
|
# nobody measured. Whether it is known is reported alongside it.
|
|
security = 0
|
|
security_known = True
|
|
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_known = False
|
|
else:
|
|
security_known = False
|
|
|
|
sizes = dnf_download_sizes() if packages else {}
|
|
for package in packages:
|
|
if package["name"] in sizes:
|
|
package["bytes"] = sizes[package["name"]]
|
|
|
|
packages.sort(key=lambda item: item["name"])
|
|
source = {"available": True, "count": len(packages), "packages": packages,
|
|
"securityCount": security, "securityKnown": security_known, "error": ""}
|
|
# Only when every pending package was priced. A partial total reads as the
|
|
# whole download and would understate it, which is the direction that
|
|
# surprises somebody on a metered connection.
|
|
if packages and all("bytes" in package for package in packages):
|
|
source["downloadBytes"] = sum(package["bytes"] for package in packages)
|
|
return source
|
|
|
|
|
|
def human_bytes(text: str) -> int:
|
|
"""flatpak's "36.2 MB" back into a number, or 0 when it is not one.
|
|
|
|
flatpak has no machine-readable size: its --json output omits the column
|
|
entirely, so the rendered string is the only source there is. Parsed with
|
|
the decimal units flatpak actually prints, and zero on anything else.
|
|
"""
|
|
match = re.match(r"^\s*([0-9]+(?:\.[0-9]+)?)\s*([kMGT]?B)\s*$", text)
|
|
if not match:
|
|
return 0
|
|
scale = {"B": 1, "kB": 10 ** 3, "MB": 10 ** 6, "GB": 10 ** 9, "TB": 10 ** 12}
|
|
return int(float(match.group(1)) * scale[match.group(2)])
|
|
|
|
|
|
def flatpak_updates() -> dict:
|
|
if not shutil.which("flatpak"):
|
|
return {"available": False, "count": 0, "applications": [], "error": ""}
|
|
result = run(["flatpak", "remote-ls", "--updates",
|
|
"--columns=application,version,origin,download-size"], timeout=120)
|
|
# An unreachable remote exits non-zero and prints nothing usable. Reading
|
|
# that as an empty list would report every application as current.
|
|
if result.returncode != 0:
|
|
return {"available": True, "count": 0, "applications": [],
|
|
"error": _refusal(result, "The application list could not be read.")}
|
|
applications = []
|
|
for line in result.stdout.splitlines():
|
|
parts = [part.strip() for part in line.split("\t")]
|
|
if not parts or not parts[0]:
|
|
continue
|
|
application = {"id": parts[0],
|
|
"version": parts[1] if len(parts) > 1 else "",
|
|
"origin": parts[2] if len(parts) > 2 else ""}
|
|
size = human_bytes(parts[3]) if len(parts) > 3 else 0
|
|
if size:
|
|
application["bytes"] = size
|
|
applications.append(application)
|
|
source = {"available": True, "count": len(applications),
|
|
"applications": applications, "error": ""}
|
|
if applications and all("bytes" in application for application in applications):
|
|
source["downloadBytes"] = sum(application["bytes"] for application in applications)
|
|
return source
|
|
|
|
|
|
def strip_markup(text: str) -> str:
|
|
"""fwupd release notes are a small AppStream XML dialect, not prose.
|
|
|
|
Paragraphs and list items become lines; everything else is dropped. Doing
|
|
this here rather than in the page keeps the helper's answer plain text, the
|
|
same shape the dnf and flatpak paths return.
|
|
"""
|
|
if not text.strip():
|
|
return ""
|
|
text = re.sub(r"</p>|</li>", "\n", text)
|
|
text = re.sub(r"<li>", "• ", text)
|
|
text = re.sub(r"<[^>]+>", "", text)
|
|
lines = [line.strip() for line in text.splitlines()]
|
|
return "\n".join(line for line in lines if line)
|
|
|
|
|
|
# fwupd reports "nothing to do" through the same Error channel it uses for
|
|
# failures. These are not failures, and calling them one would put a red row on
|
|
# a machine whose firmware is simply current.
|
|
FWUPD_NOTHING_PENDING = ("no updates available", "no updatable devices",
|
|
"no supported devices")
|
|
|
|
|
|
def firmware_updates() -> dict:
|
|
if not shutil.which("fwupdmgr"):
|
|
return {"available": False, "count": 0, "devices": [], "error": ""}
|
|
result = run(["fwupdmgr", "get-updates", "--json"], timeout=120)
|
|
# The exit code is not the signal here: `fwupdmgr --json` exits 0 even when
|
|
# it failed and reports the failure inside the payload instead (verified
|
|
# against fwupd on this machine). An answer is a payload carrying Devices or
|
|
# Error; anything else -- including the empty output a crashed fwupdmgr
|
|
# leaves -- is not an answer. This used to be `result.stdout or "{}"`, which
|
|
# turned "printed nothing" into "no firmware updates".
|
|
try:
|
|
payload = json.loads(result.stdout or "")
|
|
except json.JSONDecodeError:
|
|
payload = None
|
|
if not isinstance(payload, dict) or ("Devices" not in payload and "Error" not in payload):
|
|
return {"available": True, "count": 0, "devices": [],
|
|
"error": _refusal(result, "The firmware list could not be read.")}
|
|
failure = payload.get("Error")
|
|
if isinstance(failure, dict):
|
|
message = str(failure.get("Message", "")).strip()
|
|
if message and message.lower() not in FWUPD_NOTHING_PENDING:
|
|
return {"available": True, "count": 0, "devices": [], "error": message[:200]}
|
|
|
|
devices = []
|
|
for device in payload.get("Devices", []):
|
|
releases = device.get("Releases", [])
|
|
entry = {
|
|
"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", [])),
|
|
}
|
|
# fwupd already has the vendor's release notes in hand, so they are
|
|
# kept here rather than re-fetched: the changelog for firmware costs
|
|
# nothing beyond the scan that found the update.
|
|
notes = strip_markup(str(releases[0].get("Description", ""))) if releases else ""
|
|
if notes:
|
|
entry["changelog"] = notes[:CHANGELOG_LIMIT]
|
|
devices.append(entry)
|
|
return {"available": True, "count": len(devices), "devices": devices, "error": ""}
|
|
|
|
|
|
def automatic_state() -> dict:
|
|
flatpak_timer = run(["systemctl", "--user", "is-enabled", FLATPAK_TIMER], timeout=20)
|
|
dnf_timer = run(["systemctl", "is-enabled", DNF_TIMER], timeout=20)
|
|
return {
|
|
"flatpakEnabled": flatpak_timer.stdout.strip() == "enabled",
|
|
"flatpakAvailable": flatpak_timer.stdout.strip() not in ("", "not-found"),
|
|
# Offered when the timer exists, reported as unavailable when it does
|
|
# not. See set_auto_dnf for what enabling it actually does -- it
|
|
# downloads, it does not install.
|
|
"dnfAutomaticEnabled": dnf_timer.stdout.strip() == "enabled",
|
|
"dnfAutomaticAvailable": dnf_timer.stdout.strip() not in ("", "not-found"),
|
|
}
|
|
|
|
|
|
def probe_source(probe, empty: dict) -> dict:
|
|
"""One source's answer, or its own failure -- never the other two's.
|
|
|
|
The three fail independently, which is the whole reason they are counted
|
|
separately. A dnf that times out raises out of `run`, and left unhandled it
|
|
would abort the entire check and blank the firmware and application lists
|
|
along with it.
|
|
"""
|
|
try:
|
|
return probe()
|
|
except BoundaryError as error:
|
|
return {**empty, "available": True, "count": 0, "error": str(error)}
|
|
|
|
|
|
# What each source looks like with nothing in it. Shared by the failure path and
|
|
# by a cache written before a field existed, so both produce the same shape.
|
|
EMPTY_SOURCES = {
|
|
"dnf": {"packages": [], "securityCount": 0, "securityKnown": False},
|
|
"flatpak": {"applications": []},
|
|
"firmware": {"devices": []},
|
|
}
|
|
|
|
|
|
def check() -> dict:
|
|
previous = read_cache()
|
|
now = int(time.time())
|
|
payload = {
|
|
"dnf": probe_source(dnf_updates, EMPTY_SOURCES["dnf"]),
|
|
"flatpak": probe_source(flatpak_updates, EMPTY_SOURCES["flatpak"]),
|
|
"firmware": probe_source(firmware_updates, EMPTY_SOURCES["firmware"]),
|
|
}
|
|
for name, source in payload.items():
|
|
# A source stamps its own clock only when it answered. A check that
|
|
# failed must not make the count beside it look freshly verified --
|
|
# that is exactly how a failed check came to read as "Up to date".
|
|
source["checkedAt"] = (
|
|
int(previous.get(name, {}).get("checkedAt", 0)) if source.get("error") else now
|
|
)
|
|
failed = any(source.get("error") for source in payload.values())
|
|
# The overall stamp is the last time every source answered. Carried forward
|
|
# rather than refreshed on a partial failure, so "Checked 2 minutes ago"
|
|
# never describes a check that did not complete.
|
|
payload["checkedAt"] = int(previous.get("checkedAt", 0)) if failed else now
|
|
write_cache(payload)
|
|
return payload
|
|
|
|
|
|
def cached_source(cached: dict, name: str) -> dict:
|
|
"""A stored source, filled out to the shape this version expects.
|
|
|
|
A cache written before per-source errors and stamps existed carries neither
|
|
field; defaulting them here means an old cache reads as "nothing recorded"
|
|
instead of missing a key at the surface.
|
|
"""
|
|
empty = {"available": True, "count": 0, "error": "", "checkedAt": 0,
|
|
**EMPTY_SOURCES[name]}
|
|
source = cached.get(name)
|
|
return {**empty, **source} if isinstance(source, dict) else empty
|
|
|
|
|
|
def snapshot() -> dict:
|
|
cached = read_cache()
|
|
return {
|
|
"dnf": cached_source(cached, "dnf"),
|
|
"flatpak": cached_source(cached, "flatpak"),
|
|
"firmware": cached_source(cached, "firmware"),
|
|
# 0 means never checked, which the page says rather than showing a
|
|
# confident "0 updates" it has no basis for. It only ever holds the time
|
|
# of a check where every source answered.
|
|
"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, target: str = "") -> dict:
|
|
if source == "flatpak":
|
|
if not shutil.which("flatpak"):
|
|
raise BoundaryError("Flatpak is not installed.")
|
|
command = ["flatpak", "update", "-y", "--noninteractive"]
|
|
if target:
|
|
# One application, by ID. Checked against the IDs the last scan
|
|
# actually found rather than passed through: this is the only verb
|
|
# that takes a name from the page, and the page is not the
|
|
# authority on what is pending.
|
|
pending = {str(entry.get("id", ""))
|
|
for entry in read_cache().get("flatpak", {}).get("applications", [])}
|
|
if target not in pending:
|
|
raise BoundaryError("That application does not have an update waiting.")
|
|
command.append(target)
|
|
result = run(command, 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 dnf_changelog(name: str) -> dict:
|
|
"""The best text dnf5 will actually give for one pending package.
|
|
|
|
Two sources, in the order a person cares about them. An advisory says why
|
|
the update exists and what it fixes, which is the answer when there is one.
|
|
Failing that, the rpm changelog DELTA -- `--upgrades` prints only entries
|
|
newer than what is installed, which is exactly the question being asked and
|
|
not the package's whole history.
|
|
|
|
Plenty of packages have neither. Third-party repos routinely ship with no
|
|
changelog at all, and dnf5 answers that with a header and nothing under it.
|
|
Saying so is the honest result, not a failure.
|
|
"""
|
|
advisory = run(["dnf5", "advisory", "info", "--json", "--updates",
|
|
f"--contains-pkgs={name}"], timeout=180)
|
|
if advisory.returncode == 0:
|
|
try:
|
|
entries = json.loads(advisory.stdout or "[]")
|
|
except json.JSONDecodeError:
|
|
entries = []
|
|
blocks = []
|
|
for entry in entries if isinstance(entries, list) else []:
|
|
heading = " · ".join(part for part in (
|
|
str(entry.get("Name", "")).strip(),
|
|
str(entry.get("Type", "")).strip().title(),
|
|
str(entry.get("Severity", "")).strip(),
|
|
) if part)
|
|
body = "\n".join(part for part in (
|
|
str(entry.get("Title", "")).strip(),
|
|
str(entry.get("Description", "")).strip(),
|
|
) if part)
|
|
if heading or body:
|
|
blocks.append((heading + "\n" + body).strip())
|
|
if blocks:
|
|
return {"kind": "advisory", "text": "\n\n".join(blocks)[:CHANGELOG_LIMIT]}
|
|
|
|
result = run(["dnf5", "changelog", "--upgrades", name], timeout=180)
|
|
if result.returncode == 0:
|
|
# dnf5 prints "Listing only new changelogs..." and "Changelogs for
|
|
# <nevra>" before the entries. Both are dnf talking about itself.
|
|
lines = [line for line in result.stdout.splitlines()
|
|
if not line.startswith(("Listing only ", "Changelogs for "))]
|
|
text = "\n".join(lines).strip()
|
|
if text:
|
|
return {"kind": "changelog", "text": text[:CHANGELOG_LIMIT]}
|
|
|
|
return {"kind": "none", "text": "This package publishes no changelog for the update."}
|
|
|
|
|
|
def flatpak_changelog(name: str) -> dict:
|
|
"""Whatever the remote already has cached, and nothing more.
|
|
|
|
`flatpak remote-info --log` without --cached is an ostree history walk
|
|
against the network, which is far too much work for an expander somebody
|
|
clicked. With --cached it answers from metadata already on disk, and most
|
|
remotes have nothing there -- Flathub's commit history is not part of the
|
|
summary. Absence is reported as absence.
|
|
"""
|
|
origin = ""
|
|
for entry in read_cache().get("flatpak", {}).get("applications", []):
|
|
if str(entry.get("id", "")) == name:
|
|
origin = str(entry.get("origin", ""))
|
|
break
|
|
if not origin:
|
|
listed = run(["flatpak", "list", "--app", "--columns=application,origin"], timeout=60)
|
|
for line in listed.stdout.splitlines():
|
|
parts = [part.strip() for part in line.split("\t")]
|
|
if len(parts) > 1 and parts[0] == name:
|
|
origin = parts[1]
|
|
break
|
|
if not origin or not NAME_PATTERN.match(origin):
|
|
return {"kind": "none", "text": "This application publishes no release notes."}
|
|
|
|
result = run(["flatpak", "remote-info", "--cached", "--log", origin, name], timeout=90)
|
|
if result.returncode == 0:
|
|
history = result.stdout.partition("History:")[2].strip()
|
|
if history:
|
|
return {"kind": "changelog", "text": history[:CHANGELOG_LIMIT]}
|
|
return {"kind": "none", "text": "This application publishes no release notes."}
|
|
|
|
|
|
def firmware_changelog(name: str) -> dict:
|
|
for device in read_cache().get("firmware", {}).get("devices", []):
|
|
if str(device.get("name", "")) == name:
|
|
notes = str(device.get("changelog", "")).strip()
|
|
if notes:
|
|
return {"kind": "changelog", "text": notes[:CHANGELOG_LIMIT]}
|
|
break
|
|
return {"kind": "none", "text": "This firmware update ships no release notes."}
|
|
|
|
|
|
def changelog(source: str, name: str) -> dict:
|
|
if not NAME_PATTERN.match(name):
|
|
raise BoundaryError("That is not a name this machine would have produced.")
|
|
if source == "dnf":
|
|
if not shutil.which("dnf5"):
|
|
raise BoundaryError("dnf is not installed.")
|
|
result = dnf_changelog(name)
|
|
elif source == "flatpak":
|
|
if not shutil.which("flatpak"):
|
|
raise BoundaryError("Flatpak is not installed.")
|
|
result = flatpak_changelog(name)
|
|
elif source == "firmware":
|
|
result = firmware_changelog(name)
|
|
else:
|
|
raise BoundaryError("That is not an update source.")
|
|
return {"source": source, "name": name, **result, "error": ""}
|
|
|
|
|
|
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, DNF_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
|
|
# Changelogs answer on their own, not folded into a snapshot: this is
|
|
# the one verb whose reply is about a single package, and putting it
|
|
# inside the state blob would make every reader guess which package it
|
|
# was talking about.
|
|
if len(arguments) == 3 and arguments[0] == "changelog":
|
|
print(json.dumps(changelog(arguments[1], arguments[2]),
|
|
separators=(",", ":")))
|
|
return 0
|
|
if len(arguments) in (2, 3) and arguments[0] == "apply":
|
|
target = arguments[2] if len(arguments) == 3 else ""
|
|
if target and arguments[1] != "flatpak":
|
|
raise BoundaryError("Only applications can be updated one at a time.")
|
|
outcome = apply(arguments[1], target)
|
|
# 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], "target": target, **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 | history | "
|
|
"apply dnf|flatpak|firmware [app-id] | "
|
|
"changelog dnf|flatpak|firmware NAME | "
|
|
"set-auto-flatpak true|false | set-auto-dnf true|false")
|
|
except BoundaryError as error:
|
|
# A failed changelog answers in the changelog's own shape. Returning a
|
|
# whole state blob here would hand the caller a payload with no text
|
|
# field at all, which reads as "no changelog" rather than as a refusal.
|
|
if arguments[:1] == ["changelog"]:
|
|
print(json.dumps({"source": arguments[1] if len(arguments) > 1 else "",
|
|
"name": arguments[2] if len(arguments) > 2 else "",
|
|
"kind": "none", "text": "", "error": str(error)},
|
|
separators=(",", ":")))
|
|
return 0
|
|
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:]))
|