Harden Panama doctor probes

2026-08-18 09:00:35 -04:00
parent d58c431199
commit 0ca83f74c7
6 changed files with 147 additions and 18 deletions
@@ -54,20 +54,22 @@ class DoctorConfig:

    @property
    def command_env(self) -> dict[str, str]:
-        environment = dict(os.environ)
-        environment["PATH"] = self.path
-        environment["HOME"] = str(self.home)
-        environment["XDG_CONFIG_HOME"] = str(self.config_home)
-        environment["XDG_STATE_HOME"] = str(self.state_home)
-        environment["XDG_RUNTIME_DIR"] = str(self.runtime_dir)
-        environment.pop("PANAMA_HOME_ASSISTANT_URL", None)
-        environment.pop("PANAMA_HOME_ASSISTANT_TOKEN", None)
+        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"]
+    state: Literal["ok", "missing", "timeout", "failed", "unavailable"]
    stdout: str = ""


@@ -91,6 +93,52 @@ SYSTEMCTL_COMMANDS = {
 PROCESS_NAMES = ("quickshell", "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.hyprpaper": "Hyprpaper",
+    "desktop.hypridle": "Hypridle",
+    "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:
@@ -119,6 +167,8 @@ def run_command(command: tuple[str, ...], config: DoctorConfig, cwd: Path | None
        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)
@@ -156,7 +206,7 @@ def service_check(check_id: str, title: str, service: str, config: DoctorConfig,
    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 == "missing":
+    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)

@@ -315,7 +365,10 @@ def check_processes(config: DoctorConfig) -> Check:
    for name in PROCESS_NAMES:
        result = run_command(("pgrep", "-u", str(os.getuid()), "-x", name), config)
        if result.state == "ok":
-            counts.append(sum(line.isdecimal() for line in result.stdout.splitlines()))
+            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:
@@ -332,10 +385,19 @@ def check_caffeine(config: DoctorConfig) -> Check:
    if result.state != "ok":
        return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe is unavailable.")
    uid = str(os.getuid())
-    inhibitors = sum(
-        1 for line in result.stdout.splitlines()
-        if (parts := line.split()) and len(parts) >= 8 and parts[0] == "Panama" and parts[1] == uid and parts[3].isdecimal() and parts[-2:] == ["Caffeine", "block"]
-    )
+    inhibitors = 0
+    malformed = False
+    for line in result.stdout.splitlines():
+        parts = line.split()
+        relevant = len(parts) >= 2 and parts[0] == "Panama" and parts[1] == uid and "Caffeine" in parts
+        if not relevant:
+            continue
+        if len(parts) >= 8 and parts[3].isdecimal() and parts[-2:] == ["Caffeine", "block"]:
+            inhibitors += 1
+        else:
+            malformed = True
+    if malformed:
+        return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.")
    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:
@@ -362,6 +424,14 @@ def context_versions(config: DoctorConfig) -> list[dict[str, str]]:
    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),
@@ -372,15 +442,28 @@ def collect_checks(config: DoctorConfig) -> list[Check]:
    }
    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = {check_id: executor.submit(probes[check_id]) for check_id in CHECK_ORDER}
-        return [futures[check_id].result() 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]:
-    checks = collect_checks(config)
+    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"
-    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": context_versions(config)}, "checks": [check_json(check) for check in checks]}
+    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 main(argv: list[str]) -> int: