Gabriel Brown
4910bd659c
xdg-desktop-portal-hyprland 1.4.1 segfaults in its own shutdown path: exit() runs the CPortalManager destructor, which tears down wlr-foreign-toplevel handles and xdg-output proxies after the display has gone, and dies in wl_map_insert_at. systemd restarts it, so the desktop keeps working and the only symptom is screen sharing dropping mid-session and the occasional document-portal casualty. Sixty crashes in one day left no trace anywhere a person would look. It took reading coredumpctl to find, and nobody reads coredumpctl on a desktop that appears to be working -- which is exactly the shape of problem the health page exists for. On this machine the new check reported twenty-eight in the last day the moment it was wired up. Reported rather than repaired, and it says so. There is nothing to restart that systemd has not already restarted, and no newer build in the Hyprland COPR to install: the fix is upstream. A button that did nothing would be worse than a sentence that explains. Below five in a day stays green, because an upstream bug that only bites on shutdown will occasionally bite and a health page that is never green teaches people to ignore it.
1141 lines
56 KiB
Python
Executable File
1141 lines
56 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
|
|
"""Redacted diagnostics and bounded repairs for Panama-owned functionality."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ctypes
|
|
import errno
|
|
import json
|
|
import os
|
|
import re
|
|
import secrets
|
|
import signal
|
|
import shutil
|
|
import time
|
|
import subprocess
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from types import MappingProxyType
|
|
from typing import Callable, Literal
|
|
|
|
Status = Literal["ok", "warning", "error", "unconfigured"]
|
|
Group = Literal["desktop-foundation", "input-media", "integrations", "panama-tools"]
|
|
ActionKind = Literal["repair", "open", "instructions"]
|
|
InhibitorRow = tuple[str, str, str, str, str, str, str, str]
|
|
|
|
AT_FDCWD = -100
|
|
RENAME_NOREPLACE = 1
|
|
RENAME_EXCHANGE = 2
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Action:
|
|
kind: ActionKind
|
|
label: str
|
|
confirm: bool = False
|
|
# Only authored Settings page IDs and instruction IDs are allowed here.
|
|
# Repair commands never receive a caller-controlled target.
|
|
target: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Check:
|
|
id: str
|
|
group: Group
|
|
title: str
|
|
status: Status
|
|
detail: str
|
|
action: Action | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class DoctorConfig:
|
|
root: Path
|
|
home: Path
|
|
config_home: Path
|
|
state_home: Path
|
|
runtime_dir: Path
|
|
path: str
|
|
timeout: float
|
|
# Defaults to this file's own directory, where its sibling helpers
|
|
# (panama-action, panama-brightness, calendar-agenda) actually live.
|
|
scripts_dir: Path = Path(__file__).resolve().parent
|
|
# Repair actions (service restarts, the Quickshell restart-shell action)
|
|
# can legitimately run longer than a quick health-check probe -- a
|
|
# Quickshell restart alone waits for the old process to exit, the new one
|
|
# to start, and settle. Reusing `timeout` here would kill a slow-but-
|
|
# successful repair and report it as failed even though a following
|
|
# health scan would show everything recovered. See run_repair_command.
|
|
repair_timeout: float = 15.0
|
|
# The kernel's mount table, injectable for the same reason every other path
|
|
# here is: so the contract can exercise both a mounted and an unmounted
|
|
# document portal without needing root and a real fuse mount to do it.
|
|
mountinfo: Path = Path("/proc/self/mountinfo")
|
|
|
|
@property
|
|
def command_env(self) -> dict[str, str]:
|
|
environment = {
|
|
"PATH": self.path,
|
|
"HOME": str(self.home),
|
|
"XDG_CONFIG_HOME": str(self.config_home),
|
|
"XDG_STATE_HOME": str(self.state_home),
|
|
"XDG_RUNTIME_DIR": str(self.runtime_dir),
|
|
}
|
|
for name in PROBE_ENVIRONMENT_KEYS:
|
|
if value := os.environ.get(name):
|
|
environment[name] = value
|
|
return environment
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CommandResult:
|
|
state: Literal["ok", "missing", "timeout", "failed", "unavailable"]
|
|
stdout: str = ""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepairResult:
|
|
check_id: str
|
|
accepted: bool
|
|
exit_code: int
|
|
message: str
|
|
|
|
def as_json(self) -> dict[str, object]:
|
|
return {
|
|
"schemaVersion": 1,
|
|
"checkId": self.check_id,
|
|
"accepted": self.accepted,
|
|
"exitCode": self.exit_code,
|
|
"message": self.message,
|
|
}
|
|
|
|
|
|
CHECK_ORDER = (
|
|
"desktop.hyprland", "desktop.quickshell", "desktop.notifications", "desktop.portals",
|
|
"desktop.document-portal", "desktop.portal-stability",
|
|
"desktop.hyprpaper", "desktop.hypridle", "desktop.hyprlock", "desktop.vicinae", "input.pipewire",
|
|
"input.clipboard", "input.wallpaper", "input.capture", "input.ocr", "input.brightness",
|
|
"integration.nextcloud", "integration.rustdesk", "integration.kdeconnect", "integration.bluebubbles",
|
|
"integration.home-assistant", "integration.calendar", "panama.updates", "panama.runtime-links", "panama.vicinae-commands",
|
|
"panama.selected-terminal", "panama.selected-launcher", "panama.processes", "panama.caffeine",
|
|
)
|
|
|
|
SYSTEMCTL_COMMANDS = {
|
|
"hyprpaper": ("systemctl", "--user", "is-active", "--quiet", "hyprpaper.service"),
|
|
"hypridle": ("systemctl", "--user", "is-active", "--quiet", "hypridle.service"),
|
|
"vicinae": ("systemctl", "--user", "is-active", "--quiet", "vicinae.service"),
|
|
"pipewire": ("systemctl", "--user", "is-active", "--quiet", "pipewire.service"),
|
|
# System-scope, not --user: RustDesk ships an enabled *system* service
|
|
# (`rustdesk --service`, root-owned) that spawns the session --server and
|
|
# --tray on its own -- see autostart.lua's comment on why Panama doesn't
|
|
# start it a second time. Querying --user here always reports inactive,
|
|
# since no such user-scope unit exists, regardless of whether the real
|
|
# service is running.
|
|
"rustdesk": ("systemctl", "is-active", "--quiet", "rustdesk.service"),
|
|
}
|
|
REPAIR_COMMANDS = MappingProxyType({
|
|
"desktop.hyprpaper": ("systemctl", "--user", "restart", "hyprpaper.service"),
|
|
"desktop.hypridle": ("systemctl", "--user", "restart", "hypridle.service"),
|
|
"desktop.vicinae": ("systemctl", "--user", "restart", "vicinae.service"),
|
|
"desktop.quickshell": ("panama-action", "restart-shell"),
|
|
# Restarting is the whole repair: the drop-in in config/copy clears the
|
|
# stale mountpoint on the way in, so a start is all it takes to remount.
|
|
"desktop.document-portal": ("systemctl", "--user", "restart", "xdg-document-portal.service"),
|
|
})
|
|
RUNTIME_LINK_TARGETS = (
|
|
("hypr", Path("config/dot/hypr")),
|
|
("quickshell", Path("config/dot/quickshell")),
|
|
("uwsm", Path("config/dot/uwsm")),
|
|
("vicinae", Path("config/dot/vicinae")),
|
|
)
|
|
REPAIR_IDS = frozenset((*REPAIR_COMMANDS.keys(), "panama.runtime-links", "panama.vicinae-commands", "panama.caffeine"))
|
|
PROCESS_NAMES = ("vicinae", "hyprpaper", "hypridle")
|
|
VERSION_PATTERN = re.compile(r"\b\d+(?:\.\d+){0,3}(?:[-+._][A-Za-z0-9._-]+)?\b")
|
|
REVISION_PATTERN = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE)
|
|
PROBE_ENVIRONMENT_KEYS = (
|
|
"LANG",
|
|
"LC_ALL",
|
|
"LC_CTYPE",
|
|
"TZ",
|
|
"DBUS_SESSION_BUS_ADDRESS",
|
|
"WAYLAND_DISPLAY",
|
|
"DISPLAY",
|
|
"XAUTHORITY",
|
|
"PANAMA_DOCTOR_FIXTURE_STOPPED",
|
|
"PANAMA_DOCTOR_FIXTURE_PROCESSES",
|
|
"PANAMA_DOCTOR_FIXTURE_BUS",
|
|
"PANAMA_DOCTOR_FIXTURE_QS",
|
|
"PANAMA_DOCTOR_FIXTURE_QS_VERSION",
|
|
"PANAMA_DOCTOR_FIXTURE_BLUEBUBBLES",
|
|
"PANAMA_DOCTOR_FIXTURE_CALENDAR",
|
|
"PANAMA_DOCTOR_FIXTURE_BRIGHTNESS",
|
|
"PANAMA_DOCTOR_FIXTURE_CAFFEINE",
|
|
)
|
|
CHECK_TITLES = {
|
|
"desktop.hyprland": "Hyprland",
|
|
"desktop.quickshell": "Quickshell",
|
|
"desktop.notifications": "Notifications",
|
|
"desktop.portals": "Desktop portals",
|
|
"desktop.document-portal": "Document portal",
|
|
"desktop.portal-stability": "Portal stability",
|
|
"desktop.hyprpaper": "Hyprpaper",
|
|
"desktop.hypridle": "Hypridle",
|
|
"desktop.hyprlock": "Lock screen",
|
|
"desktop.vicinae": "Vicinae",
|
|
"input.pipewire": "PipeWire",
|
|
"input.clipboard": "Clipboard",
|
|
"input.wallpaper": "Wallpaper",
|
|
"input.capture": "Capture",
|
|
"input.ocr": "OCR",
|
|
"input.brightness": "External monitor brightness",
|
|
"integration.nextcloud": "Nextcloud",
|
|
"integration.rustdesk": "RustDesk",
|
|
"integration.kdeconnect": "KDE Connect",
|
|
"integration.bluebubbles": "BlueBubbles",
|
|
"integration.home-assistant": "Home Assistant",
|
|
"integration.calendar": "Calendar",
|
|
"panama.runtime-links": "Panama runtime links",
|
|
"panama.vicinae-commands": "Panama commands",
|
|
"panama.selected-terminal": "Selected terminal",
|
|
"panama.selected-launcher": "Selected launcher",
|
|
"panama.processes": "Panama processes",
|
|
"panama.caffeine": "Caffeine inhibitor",
|
|
}
|
|
|
|
|
|
def environment_path(name: str, default: Path) -> Path:
|
|
value = os.environ.get(name)
|
|
return Path(value).expanduser() if value else default
|
|
|
|
|
|
def config_from_environment() -> DoctorConfig:
|
|
home = environment_path("PANAMA_DOCTOR_HOME", Path.home())
|
|
config_home = environment_path("PANAMA_DOCTOR_CONFIG_HOME", Path(os.environ.get("XDG_CONFIG_HOME", home / ".config")))
|
|
state_home = environment_path("PANAMA_DOCTOR_STATE_HOME", Path(os.environ.get("XDG_STATE_HOME", home / ".local/state")))
|
|
runtime_dir = environment_path("PANAMA_DOCTOR_RUNTIME_DIR", Path(os.environ.get("XDG_RUNTIME_DIR", "/run/user/0")))
|
|
root = environment_path("PANAMA_DOCTOR_ROOT", Path(__file__).resolve().parents[4])
|
|
# Sibling helpers (panama-action, panama-brightness, calendar-agenda) live
|
|
# next to this file. PATH is not a reliable way to find them -- nothing in
|
|
# this repository puts the Quickshell scripts directory on PATH -- so they
|
|
# are invoked by resolved path instead, the same way check_hyprlock already
|
|
# resolves panama-lock.
|
|
scripts_dir = environment_path("PANAMA_DOCTOR_SCRIPTS_DIR", Path(__file__).resolve().parent)
|
|
mountinfo = environment_path("PANAMA_DOCTOR_MOUNTINFO", Path("/proc/self/mountinfo"))
|
|
try:
|
|
timeout = float(os.environ.get("PANAMA_DOCTOR_TIMEOUT", "3"))
|
|
except ValueError:
|
|
timeout = 3.0
|
|
timeout = max(0.05, min(timeout, 15.0))
|
|
try:
|
|
repair_timeout = float(os.environ.get("PANAMA_DOCTOR_REPAIR_TIMEOUT", "15"))
|
|
except ValueError:
|
|
repair_timeout = 15.0
|
|
# Never shorter than the probe timeout, and bounded so a hung repair still
|
|
# gives up rather than blocking the caller indefinitely.
|
|
repair_timeout = max(timeout, min(repair_timeout, 30.0))
|
|
return DoctorConfig(root, home, config_home, state_home, runtime_dir, os.environ.get("PANAMA_DOCTOR_PATH", os.environ.get("PATH", "")), timeout, scripts_dir, repair_timeout, mountinfo)
|
|
|
|
|
|
def run_command(command: tuple[str, ...], config: DoctorConfig, cwd: Path | None = None) -> CommandResult:
|
|
"""Run an authored read-only command without reporting its unparsed output."""
|
|
try:
|
|
completed = subprocess.run(command, capture_output=True, text=True, timeout=config.timeout, check=False, env=config.command_env, cwd=cwd)
|
|
except FileNotFoundError:
|
|
return CommandResult("missing")
|
|
except subprocess.TimeoutExpired:
|
|
return CommandResult("timeout")
|
|
except OSError:
|
|
return CommandResult("unavailable")
|
|
if completed.returncode != 0:
|
|
return CommandResult("failed")
|
|
return CommandResult("ok", completed.stdout)
|
|
|
|
|
|
def run_repair_command(command: tuple[str, ...], config: DoctorConfig, cwd: Path | None = None) -> tuple[int, str]:
|
|
"""Execute one authored repair argv and retain output only for strict parsing."""
|
|
try:
|
|
completed = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=config.repair_timeout,
|
|
check=False,
|
|
env=config.command_env,
|
|
cwd=cwd,
|
|
)
|
|
except FileNotFoundError:
|
|
return 127, ""
|
|
except subprocess.TimeoutExpired:
|
|
return 124, ""
|
|
except OSError:
|
|
return 126, ""
|
|
exit_code = completed.returncode if 0 <= completed.returncode <= 255 else 1
|
|
return exit_code, completed.stdout
|
|
|
|
|
|
def executable_exists(name: str, config: DoctorConfig) -> bool:
|
|
return shutil.which(name, path=config.path) is not None
|
|
|
|
|
|
def action_json(action: Action) -> dict[str, object]:
|
|
result: dict[str, object] = {"kind": action.kind, "label": action.label, "confirm": action.confirm}
|
|
if action.target is not None:
|
|
result["target"] = action.target
|
|
return result
|
|
|
|
|
|
def check_json(check: Check) -> dict[str, object]:
|
|
result: dict[str, object] = {"id": check.id, "group": check.group, "title": check.title, "status": check.status, "detail": check.detail}
|
|
if check.action is not None:
|
|
result["action"] = action_json(check.action)
|
|
return result
|
|
|
|
|
|
def group_for(check_id: str) -> Group:
|
|
if check_id.startswith("desktop."):
|
|
return "desktop-foundation"
|
|
if check_id.startswith("input."):
|
|
return "input-media"
|
|
if check_id.startswith("integration."):
|
|
return "integrations"
|
|
return "panama-tools"
|
|
|
|
|
|
def service_check(check_id: str, title: str, service: str, config: DoctorConfig, action: Action | None = None) -> Check:
|
|
result = run_command(SYSTEMCTL_COMMANDS[service], config)
|
|
if result.state == "ok":
|
|
return Check(check_id, group_for(check_id), title, "ok", "Service is active.")
|
|
if result.state in {"missing", "unavailable"}:
|
|
return Check(check_id, group_for(check_id), title, "error", "Required system service probe is unavailable.")
|
|
return Check(check_id, group_for(check_id), title, "warning", "Service is not active.", action)
|
|
|
|
|
|
def process_check(check_id: str, title: str, process: str, config: DoctorConfig, action: Action | None = None) -> Check:
|
|
"""Like service_check, for autostarted apps with no systemd unit behind them."""
|
|
result = run_command(("pgrep", "-u", str(os.getuid()), "-x", process), config)
|
|
if result.state == "ok":
|
|
return Check(check_id, group_for(check_id), title, "ok", "Process is running.")
|
|
if result.state in {"missing", "unavailable"}:
|
|
return Check(check_id, group_for(check_id), title, "error", "Required process probe is unavailable.")
|
|
return Check(check_id, group_for(check_id), title, "warning", "Process is not running.", action)
|
|
|
|
|
|
def ipc_target(config: DoctorConfig, target: str) -> CommandResult:
|
|
result = run_command(("qs", "ipc", "show"), config)
|
|
if result.state != "ok":
|
|
return result
|
|
return CommandResult("ok") if f"target {target}" in result.stdout.splitlines() else CommandResult("failed")
|
|
|
|
|
|
def simple_ipc_check(check_id: str, title: str, target: str, config: DoctorConfig) -> Check:
|
|
result = ipc_target(config, target)
|
|
if result.state == "ok":
|
|
return Check(check_id, "input-media", title, "ok", "Panama IPC target is available.")
|
|
if result.state == "missing":
|
|
return Check(check_id, "input-media", title, "error", "Required Quickshell executable is unavailable.")
|
|
if result.state == "timeout":
|
|
return Check(check_id, "input-media", title, "warning", "Panama IPC probe timed out.")
|
|
return Check(check_id, "input-media", title, "warning", "Panama IPC target is unavailable.")
|
|
|
|
|
|
def check_hyprland(config: DoctorConfig) -> Check:
|
|
if "hyprland" in os.environ.get("XDG_CURRENT_DESKTOP", "").casefold():
|
|
return Check("desktop.hyprland", "desktop-foundation", "Hyprland", "ok", "Hyprland session detected.")
|
|
return Check("desktop.hyprland", "desktop-foundation", "Hyprland", "error", "Hyprland session is not active.")
|
|
|
|
|
|
def check_quickshell(config: DoctorConfig) -> Check:
|
|
result = run_command(("qs", "--version"), config)
|
|
repair = Action("repair", "Restart Panama", True)
|
|
if result.state == "ok" and VERSION_PATTERN.search(result.stdout):
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "ok", "Quickshell executable is available.")
|
|
if result.state == "missing":
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "error", "Required Quickshell executable is unavailable.", repair)
|
|
return Check("desktop.quickshell", "desktop-foundation", "Quickshell", "warning", "Quickshell probe returned an invalid result.", repair)
|
|
|
|
|
|
def check_notifications(config: DoctorConfig) -> Check:
|
|
result = ipc_target(config, "notifications")
|
|
return Check("desktop.notifications", "desktop-foundation", "Notifications", "ok", "Notification service is available.") if result.state == "ok" else Check("desktop.notifications", "desktop-foundation", "Notifications", "warning", "Notification service is unavailable.")
|
|
|
|
|
|
def check_portals(config: DoctorConfig) -> Check:
|
|
result = run_command(("busctl", "--user", "--no-pager", "list"), config)
|
|
if result.state == "ok" and any(line.startswith("org.freedesktop.portal.Desktop ") for line in result.stdout.splitlines()):
|
|
return Check("desktop.portals", "desktop-foundation", "Desktop portals", "ok", "Desktop portal service is available.")
|
|
detail = "Desktop portal probe timed out." if result.state == "timeout" else "Desktop portal probe is unavailable." if result.state == "missing" else "Desktop portal service is unavailable."
|
|
return Check("desktop.portals", "desktop-foundation", "Desktop portals", "warning", detail)
|
|
|
|
|
|
def check_portal_stability(config: DoctorConfig) -> Check:
|
|
"""Whether the portal backend is crash-looping.
|
|
|
|
xdg-desktop-portal-hyprland segfaults in its own shutdown path on 1.4.1:
|
|
exit() runs the CPortalManager destructor, which tears down
|
|
wlr-foreign-toplevel handles and xdg-output proxies after the display has
|
|
gone, and dies in wl_map_insert_at. systemd restarts it, so the desktop
|
|
keeps working and the only symptom is screen sharing dying mid-call and
|
|
the occasional document-portal casualty.
|
|
|
|
That combination -- harmful but self-repairing -- is exactly what a health
|
|
page is for. Sixty crashes in a day left no trace anywhere a person would
|
|
look; it took reading coredumpctl to find out, and nobody reads coredumpctl
|
|
on a desktop that appears to be working.
|
|
|
|
Reported rather than repaired. There is nothing to restart that systemd has
|
|
not already restarted, and no newer build in the Hyprland COPR to install:
|
|
the fix is upstream. Saying so plainly is more use than a button that does
|
|
nothing.
|
|
"""
|
|
result = run_command(
|
|
("coredumpctl", "list", "--no-pager", "--since", "24 hours ago",
|
|
"/usr/libexec/xdg-desktop-portal-hyprland"),
|
|
config,
|
|
)
|
|
# No coredumps at all makes coredumpctl exit non-zero, which is the good
|
|
# case and must not read as a broken probe.
|
|
if result.state == "missing":
|
|
return Check("desktop.portal-stability", "desktop-foundation", "Portal stability",
|
|
"ok", "Crash reporting is not installed, so this cannot be checked.")
|
|
if result.state not in {"ok", "failed"}:
|
|
return Check("desktop.portal-stability", "desktop-foundation", "Portal stability",
|
|
"ok", "Could not read the crash log.")
|
|
|
|
crashes = sum(
|
|
1 for line in result.stdout.splitlines()
|
|
if "xdg-desktop-portal-hyprland" in line
|
|
)
|
|
if crashes == 0:
|
|
return Check("desktop.portal-stability", "desktop-foundation", "Portal stability",
|
|
"ok", "The portal backend has not crashed today.")
|
|
|
|
# One or two in a day is the ordinary cost of an upstream bug that only
|
|
# bites on shutdown. A dozen means something is driving it, and screen
|
|
# sharing will be unreliable.
|
|
detail = (
|
|
f"The Hyprland portal backend crashed {crashes} time"
|
|
f"{'s' if crashes != 1 else ''} in the last day. It restarts itself, so "
|
|
"the desktop keeps working, but screen sharing can drop mid-session. "
|
|
"This is an upstream fault in its shutdown path, not a setting."
|
|
)
|
|
return Check("desktop.portal-stability", "desktop-foundation", "Portal stability",
|
|
"warning" if crashes >= 5 else "ok", detail)
|
|
|
|
|
|
def check_document_portal(config: DoctorConfig) -> Check:
|
|
"""Whether the flatpak document portal is actually mounted.
|
|
|
|
Deliberately not a service probe, and deliberately separate from
|
|
check_portals -- which asks about xdg-desktop-portal, a different service
|
|
that was up and healthy throughout the outage this exists for.
|
|
|
|
xdg-document-portal provides the fuse mount at $XDG_RUNTIME_DIR/doc, and
|
|
bwrap binds doc/by-app/<id> into every flatpak sandbox it builds. Without
|
|
it nothing flatpak launches -- all of them, not one, because they share it.
|
|
|
|
The shipped unit is Type=dbus with Restart=no, so after a crash nothing
|
|
restarts it; it comes back when something next calls its bus name, and that
|
|
activation can land on the dying instance's mountpoint and come up with no
|
|
mount at all. `systemctl status` says active (running) either way, with the
|
|
fusermount3 helper still sitting there as a child. So service state is
|
|
exactly the wrong question, and asking it is how this went unnoticed for
|
|
three days and two separate debugging sessions. The only honest question is
|
|
whether the mount is in the table.
|
|
|
|
Nothing already running notices, either: a sandbox needs the mount only
|
|
while it is being constructed. That is what makes it worth a check rather
|
|
than trusting you will spot it -- the symptom arrives whenever you next
|
|
open a flatpak you had not opened yet, with nothing to connect it to.
|
|
"""
|
|
expected = str(config.runtime_dir / "doc")
|
|
try:
|
|
table = config.mountinfo.read_text(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return Check("desktop.document-portal", "desktop-foundation", "Document portal", "error", "Mount table probe is unavailable.")
|
|
|
|
for line in table.splitlines():
|
|
fields = line.split()
|
|
# mountinfo carries a variable number of optional fields, terminated by
|
|
# a lone "-": the filesystem type is the field after it, and the mount
|
|
# point is always the fifth. Splitting on "-" rather than counting from
|
|
# the end is what makes this robust to that variability.
|
|
if "-" not in fields:
|
|
continue
|
|
separator = fields.index("-")
|
|
if separator < 5 or len(fields) <= separator + 1:
|
|
continue
|
|
# Mount points are octal-escaped in this file; the space is the only
|
|
# escape that can appear in a runtime directory path in practice.
|
|
if fields[4].replace("\\040", " ") != expected:
|
|
continue
|
|
if fields[separator + 1] == "fuse.portal":
|
|
return Check("desktop.document-portal", "desktop-foundation", "Document portal", "ok", "Flatpak document portal is mounted.")
|
|
|
|
return Check(
|
|
"desktop.document-portal", "desktop-foundation", "Document portal", "warning",
|
|
"Flatpak document portal is not mounted, so no flatpak can launch.",
|
|
Action("repair", "Restart the document portal"),
|
|
)
|
|
|
|
|
|
def check_hyprlock(config: DoctorConfig) -> Check:
|
|
helper = config.root / "config/dot/quickshell/scripts/panama-lock"
|
|
tracked_fallback = config.config_home / "hypr/hyprlock.conf"
|
|
generated_config = config.state_home / "panama/hyprlock.conf"
|
|
# The doctor seals PATH for every probe. Invoke the authored helper with a
|
|
# fixed system-only PATH so its bash shebang and jq dependency remain
|
|
# available without inheriting arbitrary parent executables.
|
|
result = run_command(("/usr/bin/env", "PATH=/usr/bin:/bin", str(helper), "status"), config)
|
|
|
|
if result.state != "ok":
|
|
if tracked_fallback.is_file():
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "warning", "Managed lock-screen status is unavailable; the tracked fallback remains available.")
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "error", "No usable lock-screen configuration is available.")
|
|
|
|
try:
|
|
state = json.loads(result.stdout)
|
|
generated = state["generated"]
|
|
path = state["path"]
|
|
fallback = state["fallback"]
|
|
error = state["error"]
|
|
if not isinstance(generated, bool) or not isinstance(path, str) \
|
|
or not isinstance(fallback, bool) or not isinstance(error, str):
|
|
raise ValueError
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
if tracked_fallback.is_file():
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "warning", "Managed lock-screen status is invalid; the tracked fallback remains available.")
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "error", "No usable lock-screen configuration is available.")
|
|
|
|
if generated and not fallback and generated_config.is_file():
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "ok", "Managed lock-screen configuration is available.")
|
|
if tracked_fallback.is_file():
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "warning", "The tracked lock-screen fallback is in use.")
|
|
return Check("desktop.hyprlock", "desktop-foundation", "Lock screen", "error", "No usable lock-screen configuration is available.")
|
|
|
|
|
|
def check_brightness(config: DoctorConfig) -> Check:
|
|
result = run_command((str(config.scripts_dir / "panama-brightness"), "list"), config)
|
|
instructions = Action("instructions", "View setup instructions", target="ddc-permissions")
|
|
if result.state == "timeout":
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI probe timed out.", instructions)
|
|
if result.state != "ok":
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI support is unavailable.", instructions)
|
|
try:
|
|
listing = json.loads(result.stdout)
|
|
displays, error = listing["displays"], listing["error"]
|
|
if not isinstance(displays, list) or not isinstance(error, str):
|
|
raise ValueError
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "DDC/CI probe returned an invalid result.", instructions)
|
|
if error:
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "warning", "No accessible DDC/CI bus.", instructions)
|
|
if not displays:
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "unconfigured", "No DDC/CI display is configured.")
|
|
return Check("input.brightness", "input-media", "External monitor brightness", "ok", f"{len(displays)} DDC/CI display{'s' if len(displays) != 1 else ''} available.")
|
|
|
|
|
|
def check_nextcloud(config: DoctorConfig) -> Check:
|
|
# The client's autostart entry is Title-cased ("Nextcloud.desktop") on
|
|
# current nextcloud-client packages, but that has changed case before --
|
|
# match either so a future package update doesn't silently reintroduce
|
|
# this as "autostart is not configured" again.
|
|
autostart_dir = config.config_home / "autostart"
|
|
if not any(autostart_dir.glob("[Nn]extcloud.desktop")):
|
|
return Check("integration.nextcloud", "integrations", "Nextcloud", "unconfigured", "Nextcloud autostart is not configured.")
|
|
# No systemd unit backs this (see the autostart.lua comment on why Panama
|
|
# doesn't start it as one): the client is a plain autostarted process, so
|
|
# check for the process directly instead of a systemctl service that was
|
|
# never going to exist.
|
|
return process_check("integration.nextcloud", "Nextcloud", "nextcloud", config, Action("open", "Open Nextcloud"))
|
|
|
|
|
|
def check_rustdesk(config: DoctorConfig) -> Check:
|
|
if not executable_exists("rustdesk", config):
|
|
return Check("integration.rustdesk", "integrations", "RustDesk", "unconfigured", "RustDesk is not installed.")
|
|
return service_check("integration.rustdesk", "RustDesk", "rustdesk", config, Action("open", "Open RustDesk"))
|
|
|
|
|
|
def check_kdeconnect(config: DoctorConfig) -> Check:
|
|
if not executable_exists("kdeconnect-cli", config):
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "unconfigured", "KDE Connect is not installed.")
|
|
result = run_command(("busctl", "--user", "--no-pager", "list"), config)
|
|
if result.state == "ok" and any(line.startswith("org.kde.kdeconnect ") for line in result.stdout.splitlines()):
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "ok", "KDE Connect service is available.")
|
|
return Check("integration.kdeconnect", "integrations", "KDE Connect", "warning", "KDE Connect service is unavailable.", Action("open", "Open KDE Connect"))
|
|
|
|
|
|
def check_bluebubbles(config: DoctorConfig) -> Check:
|
|
result = run_command(("flatpak", "info", "app.bluebubbles.BlueBubbles"), config)
|
|
if result.state == "ok":
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "ok", "BlueBubbles is installed.")
|
|
if result.state in {"missing", "failed"}:
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "unconfigured", "BlueBubbles is not installed.")
|
|
return Check("integration.bluebubbles", "integrations", "BlueBubbles", "warning", "BlueBubbles installation probe timed out.", Action("open", "Open BlueBubbles"))
|
|
|
|
|
|
def check_home_assistant(config: DoctorConfig) -> Check:
|
|
configured = all(name in os.environ for name in ("PANAMA_HOME_ASSISTANT_URL", "PANAMA_HOME_ASSISTANT_TOKEN"))
|
|
helper = config.config_home / "quickshell" / "scripts" / "panama-home-assistant"
|
|
if not configured:
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "unconfigured", "Home Assistant is not configured.")
|
|
if not helper.is_file():
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "warning", "Home Assistant bridge is unavailable.", Action("open", "Open Home settings", target="home-phone"))
|
|
return Check("integration.home-assistant", "integrations", "Home Assistant", "ok", "Home Assistant credentials are configured.")
|
|
|
|
|
|
def check_calendar(config: DoctorConfig) -> Check:
|
|
result = run_command((str(config.scripts_dir / "calendar-agenda"), "probe"), config)
|
|
action = Action("open", "Open Date & Time", target="datetime")
|
|
if result.state == "missing":
|
|
return Check("integration.calendar", "integrations", "Calendar", "unconfigured", "Calendar integration is not installed.")
|
|
if result.state == "timeout":
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe timed out.", action)
|
|
if result.state != "ok":
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe failed.", action)
|
|
try:
|
|
enabled_sources = json.loads(result.stdout)["enabledSources"]
|
|
if not isinstance(enabled_sources, int) or isinstance(enabled_sources, bool):
|
|
raise ValueError
|
|
except (json.JSONDecodeError, KeyError, TypeError, ValueError):
|
|
return Check("integration.calendar", "integrations", "Calendar", "warning", "Calendar probe returned an invalid result.", action)
|
|
if enabled_sources <= 0:
|
|
return Check("integration.calendar", "integrations", "Calendar", "unconfigured", "No enabled calendar source is configured.")
|
|
return Check("integration.calendar", "integrations", "Calendar", "ok", f"{enabled_sources} enabled calendar source{'s' if enabled_sources != 1 else ''} configured.")
|
|
|
|
|
|
def check_updates(config: DoctorConfig) -> Check:
|
|
"""Whether the machine is current, and whether it is running what it installed.
|
|
|
|
Two different questions with two different answers. A kernel that has been
|
|
installed but not booted into is the one people miss: everything reports
|
|
success, nothing looks wrong, and the security fix they installed last week
|
|
is sitting on disk unused. That is reported as its own state rather than
|
|
folded into "updates available".
|
|
|
|
Read from the Updates page's cache rather than by scanning: a health check
|
|
that took nine seconds of network work would make opening System Health feel
|
|
broken. A stale cache is reported as stale.
|
|
"""
|
|
cache = Path(os.environ.get("XDG_CACHE_HOME", config.home / ".cache")) / "panama" / "updates.json"
|
|
running = os.uname().release
|
|
|
|
newest = running
|
|
rpm_query = run_command(("rpm", "-q", "kernel", "--qf", "%{VERSION}-%{RELEASE}.%{ARCH}\\n"), config)
|
|
if rpm_query.state == "ok":
|
|
installed = [line.strip() for line in rpm_query.stdout.splitlines() if line.strip()]
|
|
if installed:
|
|
newest = installed[-1]
|
|
if newest != running:
|
|
return Check("panama.updates", "panama-tools", "Software updates", "warning",
|
|
f"A newer kernel is installed than the one running ({running} → {newest}). Restart to use it.",
|
|
action=Action("open", "Open Software Update"))
|
|
|
|
try:
|
|
payload = json.loads(cache.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return Check("panama.updates", "panama-tools", "Software updates", "unconfigured",
|
|
"Updates have not been checked yet.",
|
|
action=Action("open", "Open Software Update"))
|
|
|
|
checked_at = int(payload.get("checkedAt", 0))
|
|
age_days = (time.time() - checked_at) / 86400 if checked_at else 999
|
|
security = int(payload.get("dnf", {}).get("securityCount", 0))
|
|
total = sum(int(payload.get(source, {}).get("count", 0))
|
|
for source in ("dnf", "flatpak", "firmware"))
|
|
|
|
if security > 0:
|
|
return Check("panama.updates", "panama-tools", "Software updates", "warning",
|
|
f"{security} pending update{'' if security == 1 else 's'} carry a security advisory.",
|
|
action=Action("open", "Open Software Update"))
|
|
if age_days > 7:
|
|
return Check("panama.updates", "panama-tools", "Software updates", "unconfigured",
|
|
"Updates have not been checked in over a week.",
|
|
action=Action("open", "Open Software Update"))
|
|
if total > 0:
|
|
return Check("panama.updates", "panama-tools", "Software updates", "ok",
|
|
f"{total} update{'' if total == 1 else 's'} available, none carrying a security advisory.")
|
|
return Check("panama.updates", "panama-tools", "Software updates", "ok",
|
|
"Everything is current.")
|
|
|
|
|
|
def check_runtime_links(config: DoctorConfig) -> Check:
|
|
def valid_link(name: str, relative_source: Path) -> bool:
|
|
destination = config.config_home / name
|
|
source = config.root / relative_source
|
|
try:
|
|
return source.is_dir() and destination.is_symlink() \
|
|
and destination.resolve(strict=False) == source.resolve(strict=True)
|
|
except OSError:
|
|
return False
|
|
|
|
if any(not valid_link(name, relative_source) for name, relative_source in RUNTIME_LINK_TARGETS):
|
|
return Check("panama.runtime-links", "panama-tools", "Panama runtime links", "warning", "One or more Panama runtime links are unavailable.", Action("repair", "Repair runtime links"))
|
|
return Check("panama.runtime-links", "panama-tools", "Panama runtime links", "ok", "Panama runtime links are available.")
|
|
|
|
|
|
def check_vicinae_commands(config: DoctorConfig) -> Check:
|
|
source = config.root / "config/local/share/vicinae/scripts"
|
|
installed = config.home / ".local/share/vicinae/scripts/panama"
|
|
try:
|
|
linked = source.is_dir() and installed.is_symlink() \
|
|
and installed.resolve(strict=False) == source.resolve(strict=True)
|
|
except OSError:
|
|
linked = False
|
|
if linked:
|
|
return Check("panama.vicinae-commands", "panama-tools", "Panama commands", "ok", "Panama Vicinae commands are linked.")
|
|
return Check("panama.vicinae-commands", "panama-tools", "Panama commands", "warning", "Panama Vicinae commands are not linked.", Action("repair", "Repair command link"))
|
|
|
|
|
|
def executable_check(check_id: str, title: str, executable: str, config: DoctorConfig) -> Check:
|
|
if executable_exists(executable, config):
|
|
return Check(check_id, group_for(check_id), title, "ok", f"{title} executable is available.")
|
|
return Check(check_id, group_for(check_id), title, "warning", f"{title} executable is unavailable.")
|
|
|
|
|
|
def check_processes(config: DoctorConfig) -> Check:
|
|
# `qs` is both the long-running shell and every short-lived IPC client.
|
|
# Counting it with pgrep races the other parallel health probes and reports
|
|
# duplicates whenever one of them happens to call `qs ipc`. The instance
|
|
# list is the authoritative view and contains only actual shells.
|
|
quickshell = run_command(("qs", "list"), config)
|
|
if quickshell.state == "ok":
|
|
quickshell_count = sum(
|
|
line.startswith("Instance ") for line in quickshell.stdout.splitlines()
|
|
)
|
|
elif quickshell.state == "failed":
|
|
quickshell_count = 0
|
|
else:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe is unavailable.")
|
|
|
|
counts: list[int] = [quickshell_count]
|
|
for name in PROCESS_NAMES:
|
|
result = run_command(("pgrep", "-u", str(os.getuid()), "-x", name), config)
|
|
if result.state == "ok":
|
|
pids = result.stdout.splitlines()
|
|
if not pids or any(not pid.isdecimal() for pid in pids):
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe returned an invalid result.")
|
|
counts.append(len(pids))
|
|
elif result.state == "failed":
|
|
counts.append(0)
|
|
else:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Process probe is unavailable.")
|
|
if any(count > 1 for count in counts):
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "warning", "Duplicate Panama desktop processes detected.")
|
|
if counts[0] == 0:
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "error", "Quickshell process is not running.")
|
|
return Check("panama.processes", "panama-tools", "Panama processes", "ok", "Panama desktop process counts are normal.")
|
|
|
|
|
|
def check_caffeine(config: DoctorConfig) -> Check:
|
|
result = run_command(("systemd-inhibit", "--list", "--no-pager", "--no-legend"), config)
|
|
if result.state != "ok":
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe is unavailable.")
|
|
inhibitor_rows = parse_caffeine_rows(result.stdout, str(os.getuid()))
|
|
if inhibitor_rows is None:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.")
|
|
inhibitors = len(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
|
|
if inhibitors > 1:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Duplicate Panama Caffeine inhibitors detected.", Action("repair", "Release duplicate inhibitors"))
|
|
if inhibitors == 1:
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "ok", "One Panama Caffeine inhibitor is active.")
|
|
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "ok", "No Panama Caffeine inhibitor is active.")
|
|
|
|
|
|
def parse_version(result: CommandResult, pattern: re.Pattern[str] = VERSION_PATTERN) -> str:
|
|
match = pattern.search(result.stdout) if result.state == "ok" else None
|
|
return match.group(0) if match else "unavailable"
|
|
|
|
|
|
def context_versions(config: DoctorConfig) -> list[dict[str, str]]:
|
|
hyprland = run_command(("hyprctl", "version"), config)
|
|
quickshell = run_command(("qs", "--version"), config)
|
|
revision = run_command(("git", "rev-parse", "--short", "HEAD"), config, config.root)
|
|
fedora = "unavailable"
|
|
try:
|
|
match = re.search(r"^VERSION_ID=\"?([^\n\"]+)", Path("/etc/os-release").read_text(encoding="utf-8"), re.MULTILINE)
|
|
if match and re.fullmatch(r"[0-9.]+", match.group(1)):
|
|
fedora = match.group(1)
|
|
except OSError:
|
|
pass
|
|
return [{"id": "hyprland", "version": parse_version(hyprland)}, {"id": "quickshell", "version": parse_version(quickshell)}, {"id": "fedora", "version": fedora}, {"id": "panama", "version": parse_version(revision, REVISION_PATTERN)}]
|
|
|
|
|
|
def unavailable_check(check_id: str) -> Check:
|
|
return Check(check_id, group_for(check_id), CHECK_TITLES[check_id], "warning", "Diagnostic probe could not be completed.")
|
|
|
|
|
|
def unavailable_versions() -> list[dict[str, str]]:
|
|
return [{"id": name, "version": "unavailable"} for name in ("hyprland", "quickshell", "fedora", "panama")]
|
|
|
|
|
|
def collect_checks(config: DoctorConfig) -> list[Check]:
|
|
probes: dict[str, Callable[[], Check]] = {
|
|
"desktop.hyprland": lambda: check_hyprland(config), "desktop.quickshell": lambda: check_quickshell(config), "desktop.notifications": lambda: check_notifications(config), "desktop.portals": lambda: check_portals(config), "desktop.portal-stability": lambda: check_portal_stability(config), "desktop.document-portal": lambda: check_document_portal(config),
|
|
"desktop.hyprpaper": lambda: service_check("desktop.hyprpaper", "Hyprpaper", "hyprpaper", config, Action("repair", "Restart Hyprpaper")), "desktop.hypridle": lambda: service_check("desktop.hypridle", "Hypridle", "hypridle", config, Action("repair", "Restart Hypridle")), "desktop.hyprlock": lambda: check_hyprlock(config), "desktop.vicinae": lambda: service_check("desktop.vicinae", "Vicinae", "vicinae", config, Action("repair", "Restart Vicinae")), "input.pipewire": lambda: service_check("input.pipewire", "PipeWire", "pipewire", config),
|
|
"input.clipboard": lambda: simple_ipc_check("input.clipboard", "Clipboard", "clipboard", config), "input.wallpaper": lambda: simple_ipc_check("input.wallpaper", "Wallpaper", "wallpaper", config), "input.capture": lambda: simple_ipc_check("input.capture", "Capture", "capture", config), "input.ocr": lambda: executable_check("input.ocr", "OCR", "tesseract", config), "input.brightness": lambda: check_brightness(config),
|
|
"integration.nextcloud": lambda: check_nextcloud(config), "integration.rustdesk": lambda: check_rustdesk(config), "integration.kdeconnect": lambda: check_kdeconnect(config), "integration.bluebubbles": lambda: check_bluebubbles(config), "integration.home-assistant": lambda: check_home_assistant(config), "integration.calendar": lambda: check_calendar(config),
|
|
"panama.updates": lambda: check_updates(config), "panama.runtime-links": lambda: check_runtime_links(config), "panama.vicinae-commands": lambda: check_vicinae_commands(config), "panama.selected-terminal": lambda: executable_check("panama.selected-terminal", "Selected terminal", "kitty", config), "panama.selected-launcher": lambda: executable_check("panama.selected-launcher", "Selected launcher", "vicinae", config), "panama.processes": lambda: check_processes(config), "panama.caffeine": lambda: check_caffeine(config),
|
|
}
|
|
with ThreadPoolExecutor(max_workers=8) as executor:
|
|
futures = {check_id: executor.submit(probes[check_id]) for check_id in CHECK_ORDER}
|
|
checks: list[Check] = []
|
|
for check_id in CHECK_ORDER:
|
|
try:
|
|
checks.append(futures[check_id].result())
|
|
except Exception:
|
|
checks.append(unavailable_check(check_id))
|
|
return checks
|
|
|
|
|
|
def snapshot(config: DoctorConfig) -> dict[str, object]:
|
|
try:
|
|
checks = collect_checks(config)
|
|
except Exception:
|
|
checks = [unavailable_check(check_id) for check_id in CHECK_ORDER]
|
|
counts = {status: sum(check.status == status for check in checks) for status in ("ok", "warning", "error", "unconfigured")}
|
|
overall: Literal["healthy", "warning", "error"] = "error" if counts["error"] else "warning" if counts["warning"] else "healthy"
|
|
session = "hyprland" if "hyprland" in os.environ.get("XDG_CURRENT_DESKTOP", "").casefold() else "other"
|
|
try:
|
|
versions = context_versions(config)
|
|
except Exception:
|
|
versions = unavailable_versions()
|
|
return {"schemaVersion": 1, "generatedAt": datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z"), "summary": {"status": overall, "healthy": counts["ok"], "warnings": counts["warning"], "errors": counts["error"], "unconfigured": counts["unconfigured"]}, "context": {"session": session, "versions": versions}, "checks": [check_json(check) for check in checks]}
|
|
|
|
|
|
def repair_authored_command(check_id: str, config: DoctorConfig) -> RepairResult:
|
|
command = REPAIR_COMMANDS[check_id]
|
|
if check_id == "desktop.quickshell":
|
|
# panama-action is a sibling helper script, not a PATH-resolved
|
|
# executable; see check_brightness and check_calendar for the same
|
|
# resolution against the same bug.
|
|
command = (str(config.scripts_dir / command[0]), *command[1:])
|
|
exit_code, _ = run_repair_command(command, config)
|
|
message = "Repair completed. A fresh health check will verify recovery." if exit_code == 0 \
|
|
else "The authored repair command could not be completed."
|
|
return RepairResult(check_id, True, exit_code, message)
|
|
|
|
|
|
def lexical_path(path: Path) -> Path:
|
|
"""Normalize dot segments without following any filesystem symlink."""
|
|
return Path(os.path.abspath(os.fspath(path)))
|
|
|
|
|
|
def lexical_link_target(destination: Path) -> Path:
|
|
target = Path(os.readlink(destination))
|
|
return lexical_path(target if target.is_absolute() else destination.parent / target)
|
|
|
|
|
|
def renameat2(source: Path, destination: Path, flags: int) -> None:
|
|
"""Call Linux renameat2 with fixed flags selected by authored code."""
|
|
libc = ctypes.CDLL(None, use_errno=True)
|
|
function = getattr(libc, "renameat2", None)
|
|
if function is None:
|
|
raise OSError(errno.ENOSYS, "renameat2 is unavailable")
|
|
function.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
|
|
function.restype = ctypes.c_int
|
|
result = function(
|
|
AT_FDCWD,
|
|
os.fsencode(source),
|
|
AT_FDCWD,
|
|
os.fsencode(destination),
|
|
flags,
|
|
)
|
|
if result != 0:
|
|
error = ctypes.get_errno()
|
|
raise OSError(error, os.strerror(error), destination)
|
|
|
|
|
|
def rename_exchange(source: Path, destination: Path) -> None:
|
|
renameat2(source, destination, RENAME_EXCHANGE)
|
|
|
|
|
|
def rename_noreplace(source: Path, destination: Path) -> None:
|
|
renameat2(source, destination, RENAME_NOREPLACE)
|
|
|
|
|
|
def create_symlink_candidate(destination: Path, source: Path) -> Path:
|
|
"""Create one unpredictable authored sibling candidate symlink."""
|
|
for _ in range(32):
|
|
candidate = destination.with_name(
|
|
f".panama-link-{destination.name}-{os.getpid()}-{secrets.token_hex(8)}"
|
|
)
|
|
try:
|
|
os.symlink(source, candidate, target_is_directory=True)
|
|
return candidate
|
|
except FileExistsError:
|
|
continue
|
|
raise OSError("Could not allocate an authored temporary link")
|
|
|
|
|
|
def cleanup_candidate(candidate: Path) -> None:
|
|
try:
|
|
candidate.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def install_absent_symlink(destination: Path, source: Path) -> Literal["repaired", "blocked", "failed"]:
|
|
candidate = create_symlink_candidate(destination, source)
|
|
try:
|
|
try:
|
|
rename_noreplace(candidate, destination)
|
|
except FileExistsError:
|
|
return "blocked"
|
|
except OSError:
|
|
return "failed"
|
|
return "repaired"
|
|
finally:
|
|
cleanup_candidate(candidate)
|
|
|
|
|
|
def exchange_owned_symlink(
|
|
destination: Path,
|
|
source: Path,
|
|
authored_sources: frozenset[Path],
|
|
) -> Literal["repaired", "blocked", "failed"]:
|
|
"""Exchange first, then validate the exact object removed from destination."""
|
|
candidate = create_symlink_candidate(destination, source)
|
|
exchanged = False
|
|
rolled_back = False
|
|
try:
|
|
try:
|
|
rename_exchange(candidate, destination)
|
|
exchanged = True
|
|
except OSError:
|
|
return "failed"
|
|
|
|
try:
|
|
old_is_authored = candidate.is_symlink() \
|
|
and lexical_link_target(candidate) in authored_sources
|
|
except OSError:
|
|
old_is_authored = False
|
|
if old_is_authored:
|
|
cleanup_candidate(candidate)
|
|
return "repaired"
|
|
|
|
try:
|
|
rename_exchange(candidate, destination)
|
|
rolled_back = True
|
|
except OSError:
|
|
# The displaced object remains at the unpredictable candidate path;
|
|
# never unlink it when rollback could not restore ownership.
|
|
return "failed"
|
|
|
|
try:
|
|
restored_candidate_is_ours = candidate.is_symlink() \
|
|
and lexical_link_target(candidate) == source
|
|
except OSError:
|
|
restored_candidate_is_ours = False
|
|
if not restored_candidate_is_ours:
|
|
return "failed"
|
|
cleanup_candidate(candidate)
|
|
return "blocked"
|
|
finally:
|
|
if not exchanged or rolled_back:
|
|
try:
|
|
if candidate.is_symlink() and lexical_link_target(candidate) == source:
|
|
cleanup_candidate(candidate)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def repair_runtime_links(config: DoctorConfig) -> RepairResult:
|
|
root = lexical_path(config.root)
|
|
sources = [(name, lexical_path(config.root / relative_source)) for name, relative_source in RUNTIME_LINK_TARGETS]
|
|
if any(not source.is_dir() for _, source in sources):
|
|
return RepairResult("panama.runtime-links", True, 1, "Tracked Panama link destinations are unavailable.")
|
|
if any(not source.is_relative_to(root) for _, source in sources):
|
|
return RepairResult("panama.runtime-links", True, 1, "Tracked Panama link destinations are invalid.")
|
|
authored_sources = frozenset(source for _, source in sources)
|
|
|
|
try:
|
|
config.config_home.mkdir(parents=True, exist_ok=True)
|
|
except OSError:
|
|
return RepairResult("panama.runtime-links", True, 1, "Panama runtime links could not be accessed.")
|
|
|
|
blocked = False
|
|
failed = False
|
|
for name, source in sources:
|
|
destination = config.config_home / name
|
|
try:
|
|
if destination.is_symlink():
|
|
current_target = lexical_link_target(destination)
|
|
if current_target == source:
|
|
continue
|
|
if current_target not in authored_sources:
|
|
blocked = True
|
|
continue
|
|
outcome = exchange_owned_symlink(destination, source, authored_sources)
|
|
blocked = blocked or outcome == "blocked"
|
|
failed = failed or outcome == "failed"
|
|
elif destination.exists():
|
|
# A regular file or directory is user-owned unless proven
|
|
# otherwise. Report it, but never replace it.
|
|
blocked = True
|
|
else:
|
|
outcome = install_absent_symlink(destination, source)
|
|
blocked = blocked or outcome == "blocked"
|
|
failed = failed or outcome == "failed"
|
|
except OSError:
|
|
failed = True
|
|
|
|
if failed:
|
|
return RepairResult("panama.runtime-links", True, 1, "One or more Panama runtime links could not be recreated.")
|
|
if blocked:
|
|
return RepairResult("panama.runtime-links", True, 1, "A user-owned runtime path is blocking a Panama link.")
|
|
return RepairResult("panama.runtime-links", True, 0, "Panama runtime links were recreated. A fresh health check will verify them.")
|
|
|
|
|
|
def repair_vicinae_commands(config: DoctorConfig) -> RepairResult:
|
|
helper = config.root / "setup/scripts/link-vicinae-scripts"
|
|
if not helper.is_file():
|
|
return RepairResult("panama.vicinae-commands", True, 127, "The authored Vicinae link helper is unavailable.")
|
|
exit_code, _ = run_repair_command((str(helper),), config, config.root)
|
|
message = "Panama commands were relinked. A fresh health check will verify them." if exit_code == 0 \
|
|
else "Panama commands could not be relinked."
|
|
return RepairResult("panama.vicinae-commands", True, exit_code, message)
|
|
|
|
|
|
def parse_caffeine_rows(output: str, uid: str) -> list[InhibitorRow] | None:
|
|
inhibitor_rows: list[InhibitorRow] = []
|
|
for line in output.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 2 or parts[0] != "Panama" or parts[1] != uid:
|
|
continue
|
|
if len(parts) != 8:
|
|
if "Caffeine" in parts:
|
|
return None
|
|
continue
|
|
if parts[6] != "Caffeine" or parts[7] != "block":
|
|
continue
|
|
if not parts[3].isdecimal():
|
|
return None
|
|
inhibitor_rows.append(tuple(parts))
|
|
return inhibitor_rows
|
|
|
|
|
|
def close_pidfds(pidfds: list[int]) -> None:
|
|
for pidfd in pidfds:
|
|
try:
|
|
os.close(pidfd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def signal_caffeine_pidfds(
|
|
pidfds: list[int],
|
|
sender: Callable[..., None] | None = None,
|
|
) -> Literal["released", "preflight-failed", "incomplete"]:
|
|
send = sender or signal.pidfd_send_signal
|
|
for pidfd in pidfds:
|
|
try:
|
|
send(pidfd, 0, None, 0)
|
|
except (OSError, ValueError):
|
|
return "preflight-failed"
|
|
|
|
incomplete = False
|
|
for pidfd in pidfds:
|
|
try:
|
|
send(pidfd, signal.SIGTERM, None, 0)
|
|
except ProcessLookupError:
|
|
continue
|
|
except OSError as error:
|
|
if error.errno != errno.ESRCH:
|
|
incomplete = True
|
|
except ValueError:
|
|
incomplete = True
|
|
return "incomplete" if incomplete else "released"
|
|
|
|
|
|
def repair_caffeine(config: DoctorConfig) -> RepairResult:
|
|
list_command = ("systemd-inhibit", "--list", "--no-pager", "--no-legend")
|
|
list_exit, output = run_repair_command(list_command, config)
|
|
if list_exit != 0:
|
|
return RepairResult("panama.caffeine", True, list_exit, "Caffeine inhibitors could not be inspected.")
|
|
|
|
uid = str(os.getuid())
|
|
inhibitor_rows = parse_caffeine_rows(output, uid)
|
|
if inhibitor_rows is None:
|
|
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
|
|
inhibitor_pids = list(dict.fromkeys(int(row[3]) for row in inhibitor_rows))
|
|
|
|
if len(inhibitor_pids) <= 1:
|
|
return RepairResult("panama.caffeine", True, 0, "No duplicate Panama Caffeine inhibitors needed release.")
|
|
|
|
duplicates = inhibitor_pids[1:]
|
|
if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
|
|
return RepairResult("panama.caffeine", True, 1, "Safe Caffeine inhibitor release is unavailable on this system.")
|
|
|
|
pidfds: list[int] = []
|
|
try:
|
|
try:
|
|
pidfds = [os.pidfd_open(pid, 0) for pid in duplicates]
|
|
except (OSError, ValueError):
|
|
return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
|
|
|
|
second_exit, second_output = run_repair_command(list_command, config)
|
|
if second_exit != 0:
|
|
return RepairResult("panama.caffeine", True, second_exit, "Caffeine inhibitors could not be revalidated; nothing was released.")
|
|
second_rows = parse_caffeine_rows(second_output, uid)
|
|
if second_rows is None or second_rows != inhibitor_rows:
|
|
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata changed; nothing was released.")
|
|
|
|
signal_outcome = signal_caffeine_pidfds(pidfds)
|
|
if signal_outcome == "preflight-failed":
|
|
return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
|
|
if signal_outcome == "incomplete":
|
|
return RepairResult("panama.caffeine", True, 1, "One or more duplicate inhibitors could not be released.")
|
|
finally:
|
|
close_pidfds(pidfds)
|
|
return RepairResult("panama.caffeine", True, 0, "Duplicate Panama Caffeine inhibitors were released. A fresh health check will verify recovery.")
|
|
|
|
|
|
def repair(check_id: str, config: DoctorConfig) -> RepairResult:
|
|
if check_id in REPAIR_COMMANDS:
|
|
return repair_authored_command(check_id, config)
|
|
if check_id == "panama.runtime-links":
|
|
return repair_runtime_links(config)
|
|
if check_id == "panama.vicinae-commands":
|
|
return repair_vicinae_commands(config)
|
|
if check_id == "panama.caffeine":
|
|
return repair_caffeine(config)
|
|
return RepairResult(check_id, False, 2, "This health check has no authored repair.")
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
parser = argparse.ArgumentParser(description="Panama system diagnostics and bounded repairs")
|
|
output = parser.add_mutually_exclusive_group()
|
|
output.add_argument("--json", action="store_true")
|
|
output.add_argument("--summary", action="store_true")
|
|
parser.add_argument("--repair", metavar="CHECK_ID")
|
|
args = parser.parse_args(argv)
|
|
|
|
if args.repair is not None:
|
|
if args.repair not in REPAIR_IDS or args.summary:
|
|
result = RepairResult(args.repair, False, 2, "This health check has no authored repair.")
|
|
else:
|
|
try:
|
|
result = repair(args.repair, config_from_environment())
|
|
except Exception:
|
|
result = RepairResult(args.repair, True, 1, "The authored repair could not be completed.")
|
|
print(json.dumps(result.as_json(), separators=(",", ":"), sort_keys=False))
|
|
return result.exit_code
|
|
|
|
result = snapshot(config_from_environment())
|
|
if args.summary:
|
|
summary = result["summary"]
|
|
assert isinstance(summary, dict)
|
|
print(f"Panama system health: {summary['status']} ({summary['healthy']} ok, {summary['warnings']} warnings, {summary['errors']} errors, {summary['unconfigured']} unconfigured)")
|
|
else:
|
|
print(json.dumps(result, separators=(",", ":"), sort_keys=False))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv[1:]))
|