gib/Panama
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Tier 0: render what the services already decided, honestly

Browse Source 
Claude-Session: https://claude.ai/code/session_01Ms2FbjQy31TVf3CEvQhGM8
This commit is contained in:
Gabriel Brown
2026-08-25 00:24:30 -04:00
parent be0e55214b
commit 8b1205e4b8
38 changed files with 1474 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/dot/quickshell/scripts/panama-firewall
+12 -5
View File
@@ -15,7 +15,7 @@ Changes go through firewall-cmd, which is polkit-aware, so they prompt.
panama-firewall snapshot
panama-firewall zone-info ZONE
panama-firewall add-service NAME | remove-service NAME
panama-firewall add-port PORT/PROTO | remove-port PORT/PROTO
panama-firewall add-port PORT/PROTO... | remove-port PORT/PROTO...
panama-firewall set-zone INTERFACE ZONE
panama-firewall set-default-zone ZONE
"""
@@ -401,10 +401,17 @@ def main(arguments: list[str]) -> int:
name = require(SERVICE, arguments[1], "That is not a service name.")
verb = "--add-service" if arguments[0] == "add-service" else "--remove-service"
change(active_zone(), f"{verb}={name}")
elif len(arguments) == 2 and arguments[0] in ("add-port", "remove-port"):
spec = require(PORT_SPEC, arguments[1], "That is not a port.")
elif len(arguments) >= 2 and arguments[0] in ("add-port", "remove-port"):
# Several specs in one invocation, because one rule as a user sees
# it is often two as firewalld stores it: the range Fedora opens is
# a tcp range AND a udp range, and "close the range" that closed
# only the tcp half would be a lie the page had already told.
# Validated all-or-nothing first, so a bad spec at the end cannot
# leave the firewall half-changed.
specs = [require(PORT_SPEC, argument, "That is not a port.")
for argument in arguments[1:]]
verb = "--add-port" if arguments[0] == "add-port" else "--remove-port"
change(active_zone(), f"{verb}={spec}")
change(active_zone(), *[f"{verb}={spec}" for spec in specs])
elif len(arguments) == 3 and arguments[0] == "set-zone":
interface = require(INTERFACE, arguments[1], "That is not a network interface.")
zone = require(ZONE, arguments[2], "That is not a zone.")
@@ -416,7 +423,7 @@ def main(arguments: list[str]) -> int:
else:
raise BoundaryError(
"Usage: panama-firewall snapshot | zone-info ZONE | add-service NAME | "
"remove-service NAME | add-port PORT/PROTO | remove-port PORT/PROTO | "
"remove-service NAME | add-port PORT/PROTO... | remove-port PORT/PROTO... | "
"set-zone INTERFACE ZONE | set-default-zone ZONE")
except BoundaryError as error:
try:
config/dot/quickshell/scripts/panama-updates
+160 -60
View File
@@ -143,32 +143,56 @@ def dnf_download_sizes() -> dict[str, int]:
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}
return {"available": False, "count": 0, "packages": [], "securityCount": 0,
"securityKnown": True, "error": ""}
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
# 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 = 0
security_known = False
else:
security_known = False
sizes = dnf_download_sizes() if packages else {}
for package in packages:
@@ -177,7 +201,7 @@ def dnf_updates() -> dict:
packages.sort(key=lambda item: item["name"])
source = {"available": True, "count": len(packages), "packages": packages,
"securityCount": security}
"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.
@@ -202,23 +226,28 @@ def human_bytes(text: str) -> int:
def flatpak_updates() -> dict:
if not shutil.which("flatpak"):
return {"available": False, "count": 0, "applications": []}
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 = []
if result.returncode == 0:
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}
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
@@ -240,32 +269,54 @@ def strip_markup(text: str) -> str:
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": []}
return {"available": False, "count": 0, "devices": [], "error": ""}
result = run(["fwupdmgr", "get-updates", "--json"], timeout=120)
devices = []
# 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 "{}")
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)
payload = json.loads(result.stdout or "")
except json.JSONDecodeError:
pass
return {"available": True, "count": len(devices), "devices": devices}
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:
@@ -282,26 +333,75 @@ def automatic_state() -> dict:
}
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": dnf_updates(),
"flatpak": flatpak_updates(),
"firmware": firmware_updates(),
"checkedAt": int(time.time()),
"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()
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": []}),
"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.
# 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(),