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

Harden bounded Panama recovery actions

Browse Source
This commit is contained in:
Gabriel Brown
2026-08-18 10:58:48 -04:00
parent 2cc109f5e1
commit 8bfae70284
4 changed files with 403 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/dot/quickshell/scripts/panama-doctor
+113 -37
View File
@@ -8,6 +8,8 @@ import argparse
import json
import os
import re
import secrets
import signal
import shutil
import subprocess
import sys
@@ -409,8 +411,13 @@ def check_runtime_links(config: DoctorConfig) -> Check:
def check_vicinae_commands(config: DoctorConfig) -> Check:
source = config.root / "config/local/share/vicinae/scripts"
installed = config.home / ".local/share/vicinae/scripts"
if source.is_dir() and installed.is_symlink() and installed.exists():
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"))
@@ -445,20 +452,10 @@ 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.")
uid = str(os.getuid())
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:
inhibitor_pids = parse_caffeine_pids(result.stdout, str(os.getuid()))
if inhibitor_pids is None:
return Check("panama.caffeine", "panama-tools", "Caffeine inhibitor", "warning", "Caffeine inhibitor probe returned an invalid result.")
inhibitors = len(dict.fromkeys(inhibitor_pids))
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:
@@ -535,10 +532,47 @@ def repair_authored_command(check_id: str, config: DoctorConfig) -> RepairResult
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 atomic_symlink_replace(destination: Path, source: Path) -> None:
"""Install an authored sibling symlink without first removing destination."""
for _ in range(32):
temporary = destination.with_name(
f".panama-link-{destination.name}-{os.getpid()}-{secrets.token_hex(8)}"
)
created = False
try:
os.symlink(source, temporary, target_is_directory=True)
created = True
os.replace(temporary, destination)
return
except FileExistsError:
continue
finally:
if created:
try:
temporary.unlink()
except FileNotFoundError:
pass
raise OSError("Could not allocate an authored temporary link")
def repair_runtime_links(config: DoctorConfig) -> RepairResult:
sources = [(name, config.root / relative_source) for name, relative_source in RUNTIME_LINK_TARGETS]
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)
@@ -551,23 +585,26 @@ def repair_runtime_links(config: DoctorConfig) -> RepairResult:
destination = config.config_home / name
try:
if destination.is_symlink():
if destination.resolve(strict=False) == source.resolve(strict=True):
current_target = lexical_link_target(destination)
if current_target == source:
continue
destination.unlink()
destination.symlink_to(source, target_is_directory=True)
if current_target not in authored_sources:
blocked = True
continue
atomic_symlink_replace(destination, source)
elif destination.exists():
# A regular file or directory is user-owned unless proven
# otherwise. Report it, but never replace it.
blocked = True
else:
destination.symlink_to(source, target_is_directory=True)
atomic_symlink_replace(destination, source)
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 file or directory is blocking a Panama runtime link.")
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.")
@@ -581,6 +618,32 @@ def repair_vicinae_commands(config: DoctorConfig) -> RepairResult:
return RepairResult("panama.vicinae-commands", True, exit_code, message)
def parse_caffeine_pids(output: str, uid: str) -> list[int] | None:
inhibitor_pids: list[int] = []
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_pids.append(int(parts[3]))
return inhibitor_pids
def close_pidfds(pidfds: list[int]) -> None:
for pidfd in pidfds:
try:
os.close(pidfd)
except OSError:
pass
def repair_caffeine(config: DoctorConfig) -> RepairResult:
list_command = ("systemd-inhibit", "--list", "--no-pager", "--no-legend")
list_exit, output = run_repair_command(list_command, config)
@@ -588,26 +651,39 @@ def repair_caffeine(config: DoctorConfig) -> RepairResult:
return RepairResult("panama.caffeine", True, list_exit, "Caffeine inhibitors could not be inspected.")
uid = str(os.getuid())
inhibitor_pids: list[str] = []
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:
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
if parts[6] != "Caffeine" or parts[7] != "block":
continue
if not parts[3].isdecimal():
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
inhibitor_pids.append(parts[3])
parsed_pids = parse_caffeine_pids(output, uid)
if parsed_pids is None:
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata was invalid; nothing was released.")
inhibitor_pids = list(dict.fromkeys(parsed_pids))
if len(inhibitor_pids) <= 1:
return RepairResult("panama.caffeine", True, 0, "No duplicate Panama Caffeine inhibitors needed release.")
for pid in inhibitor_pids[1:]:
exit_code, _ = run_repair_command(("kill", "--", pid), config)
if exit_code != 0:
return RepairResult("panama.caffeine", True, exit_code, "A duplicate Panama Caffeine inhibitor could not be released.")
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_parsed = parse_caffeine_pids(second_output, uid)
if second_parsed is None or any(pid not in set(second_parsed) for pid in duplicates):
return RepairResult("panama.caffeine", True, 1, "Caffeine inhibitor metadata changed; nothing was released.")
try:
for pidfd in pidfds:
signal.pidfd_send_signal(pidfd, signal.SIGTERM, None, 0)
except (OSError, ValueError):
return RepairResult("panama.caffeine", True, 1, "A duplicate inhibitor changed before it could be safely released.")
finally:
close_pidfds(pidfds)
return RepairResult("panama.caffeine", True, 0, "Duplicate Panama Caffeine inhibitors were released. A fresh health check will verify recovery.")