Finish the wonderland: System told truthfully, in eight tabs instead of ten
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
@@ -15,7 +15,10 @@ 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
|
||||
@@ -30,10 +33,22 @@ 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.
|
||||
# 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
|
||||
@@ -103,6 +118,30 @@ def kernel_state() -> dict:
|
||||
}
|
||||
|
||||
|
||||
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:
|
||||
if not shutil.which("dnf5"):
|
||||
return {"available": False, "count": 0, "packages": [], "securityCount": 0}
|
||||
@@ -131,24 +170,74 @@ def dnf_updates() -> dict:
|
||||
except json.JSONDecodeError:
|
||||
security = 0
|
||||
|
||||
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"])
|
||||
return {"available": True, "count": len(packages), "packages": packages,
|
||||
"securityCount": security}
|
||||
source = {"available": True, "count": len(packages), "packages": packages,
|
||||
"securityCount": security}
|
||||
# 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": []}
|
||||
result = run(["flatpak", "remote-ls", "--updates", "--columns=application,version"],
|
||||
timeout=120)
|
||||
result = run(["flatpak", "remote-ls", "--updates",
|
||||
"--columns=application,version,origin,download-size"], 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}
|
||||
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}
|
||||
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)
|
||||
|
||||
|
||||
def firmware_updates() -> dict:
|
||||
@@ -160,13 +249,20 @@ def firmware_updates() -> dict:
|
||||
payload = json.loads(result.stdout or "{}")
|
||||
for device in payload.get("Devices", []):
|
||||
releases = device.get("Releases", [])
|
||||
devices.append({
|
||||
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)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
return {"available": True, "count": len(devices), "devices": devices}
|
||||
@@ -174,12 +270,13 @@ def firmware_updates() -> dict:
|
||||
|
||||
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)
|
||||
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"),
|
||||
# Reported, never offered: dnf-automatic is a package this machine does
|
||||
# not have, and installing software is not a settings action.
|
||||
# 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"),
|
||||
}
|
||||
@@ -226,11 +323,22 @@ def take_restore_point(reason: str) -> str:
|
||||
return result.stdout.strip() if result.returncode == 0 else ""
|
||||
|
||||
|
||||
def apply(source: str) -> dict:
|
||||
def apply(source: str, target: str = "") -> dict:
|
||||
if source == "flatpak":
|
||||
if not shutil.which("flatpak"):
|
||||
raise BoundaryError("Flatpak is not installed.")
|
||||
result = run(["flatpak", "update", "-y", "--noninteractive"], timeout=3600)
|
||||
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": ""}
|
||||
@@ -257,6 +365,115 @@ def apply(source: str) -> dict:
|
||||
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.
|
||||
|
||||
@@ -269,7 +486,7 @@ def set_auto_dnf(enabled: bool) -> None:
|
||||
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)
|
||||
result = run(["pkexec", "systemctl", *action, DNF_TIMER], timeout=120)
|
||||
if result.returncode != 0:
|
||||
raise BoundaryError(_refusal(result, "Automatic package updates could not be changed."))
|
||||
|
||||
@@ -383,13 +600,24 @@ def main(arguments: list[str]) -> int:
|
||||
check()
|
||||
print(json.dumps(snapshot(), separators=(",", ":")))
|
||||
return 0
|
||||
if len(arguments) == 2 and arguments[0] == "apply":
|
||||
outcome = apply(arguments[1])
|
||||
# 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], **outcome}
|
||||
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":
|
||||
@@ -398,9 +626,20 @@ def main(arguments: list[str]) -> int:
|
||||
set_auto_dnf(arguments[1] == "true")
|
||||
else:
|
||||
raise BoundaryError(
|
||||
"Usage: panama-updates snapshot | check | apply dnf|flatpak|firmware | "
|
||||
"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=(",", ":")))
|
||||
|
||||
Reference in New Issue
Block a user