core: locale-pinned pacman parsing, fail-closed scanners, root-helper hardening
- run_c_locale() helper (LC_ALL=C) for all pacman output parsing; fixes the kernel-modules scanner offering to delete the RUNNING kernel's modules on non-English locales, wrong -Qi sizes and the -Ss [installed] flag - SystemKernelsScanner fails closed when pacman -Qo yields no owners - taninux-helper: systemctl operands must be unit names (no paths — closes the `systemctl enable /path` escalation primitive); user/group operands reject / and ..; exec via fixed system dirs instead of PATH lookup - wine.safe_to_delete resolves symlinks/..; new_prefix_path rejects bad names - kernel screen: exact variant matching (linux-rt-lts no longer marks linux-rt and linux-lts as running) - update sources log failures to stderr instead of reporting "up to date" - drop dead bootloader.prune_argv (helper allowlist rejected its argv) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -129,14 +129,3 @@ def set_default_argv(bl: BootloaderInfo, entry) -> list[str] | None:
|
|||||||
# entry-id für bootctl ist der Dateiname des .conf-Eintrags
|
# entry-id für bootctl ist der Dateiname des .conf-Eintrags
|
||||||
entry_id = entry.name if isinstance(entry, Path) else str(entry)
|
entry_id = entry.name if isinstance(entry, Path) else str(entry)
|
||||||
return ["bootctl", "set-default", entry_id]
|
return ["bootctl", "set-default", entry_id]
|
||||||
|
|
||||||
|
|
||||||
def prune_argv(keep: int) -> list[str]:
|
|
||||||
"""ARGV zum Aufräumen alter zwischengespeicherter Pakete via `paccache`.
|
|
||||||
|
|
||||||
Meint den pacman-Cache (/var/cache/pacman/pkg) — darin liegen alte Versionen
|
|
||||||
aller Pakete, inkl. alter Kernel. `paccache -rk <keep>` entfernt alle bis auf
|
|
||||||
die `keep` jüngsten Versionen pro Paket; idiomatischer Arch-Weg.
|
|
||||||
core eskaliert nicht selbst — der Aufrufer führt das privilegiert aus.
|
|
||||||
"""
|
|
||||||
return ["paccache", "-rk", str(max(0, int(keep)))]
|
|
||||||
|
|||||||
@@ -4,17 +4,25 @@ Zentral statt in jedem Pillar einen Subprocess-Call. Cache wird gecleart
|
|||||||
nach jedem write-Aktion (install/remove)."""
|
nach jedem write-Aktion (install/remove)."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
from functools import lru_cache
|
from functools import lru_cache
|
||||||
|
|
||||||
|
|
||||||
|
def run_c_locale(argv: list[str], timeout: float | None = None) -> subprocess.CompletedProcess[str]:
|
||||||
|
"""subprocess.run mit LC_ALL=C — für alles, dessen Output-Parsing auf
|
||||||
|
englische Meldungen angewiesen ist (pacman/paru "is owned by",
|
||||||
|
"Installed Size", "[installed]" …)."""
|
||||||
|
return subprocess.run(
|
||||||
|
argv, capture_output=True, text=True, check=False, timeout=timeout,
|
||||||
|
env={**os.environ, "LC_ALL": "C"},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@lru_cache(maxsize=1)
|
@lru_cache(maxsize=1)
|
||||||
def installed_packages() -> frozenset[str]:
|
def installed_packages() -> frozenset[str]:
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
r = run_c_locale(["pacman", "-Qq"], timeout=10)
|
||||||
["pacman", "-Qq"],
|
|
||||||
capture_output=True, text=True, check=False, timeout=10,
|
|
||||||
)
|
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return frozenset()
|
return frozenset()
|
||||||
return frozenset(l.strip() for l in r.stdout.splitlines() if l.strip())
|
return frozenset(l.strip() for l in r.stdout.splitlines() if l.strip())
|
||||||
|
|||||||
@@ -14,6 +14,8 @@ import shutil
|
|||||||
import subprocess
|
import subprocess
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
|
|
||||||
|
from taninux.core.pacman import run_c_locale
|
||||||
|
|
||||||
MAX_RESULTS = 60
|
MAX_RESULTS = 60
|
||||||
|
|
||||||
_HEAD = re.compile(r"^(?P<repo>[\w.-]+)/(?P<name>\S+)\s+(?P<ver>\S+)(?P<rest>.*)$")
|
_HEAD = re.compile(r"^(?P<repo>[\w.-]+)/(?P<name>\S+)\s+(?P<ver>\S+)(?P<rest>.*)$")
|
||||||
@@ -42,9 +44,8 @@ def _search_pacman(query: str, include_aur: bool) -> list[Result]:
|
|||||||
helper = _helper() if include_aur else None
|
helper = _helper() if include_aur else None
|
||||||
argv = [helper, "-Ss", query] if helper else ["pacman", "-Ss", query]
|
argv = [helper, "-Ss", query] if helper else ["pacman", "-Ss", query]
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
# LC_ALL=C: der "[installed"-Tag im -Ss-Output ist lokalisiert.
|
||||||
argv, capture_output=True, text=True, check=False, timeout=45,
|
r = run_c_locale(argv, timeout=45)
|
||||||
)
|
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return []
|
return []
|
||||||
if r.returncode not in (0, 1):
|
if r.returncode not in (0, 1):
|
||||||
|
|||||||
@@ -110,16 +110,26 @@ def winetricks_argv(prefix: Path, *verbs: str) -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
def new_prefix_path(name: str) -> Path:
|
def new_prefix_path(name: str) -> Path:
|
||||||
|
if not name or "/" in name or ".." in name:
|
||||||
|
raise ValueError(f"invalid prefix name: {name!r}")
|
||||||
PREFIXES_DIR.mkdir(parents=True, exist_ok=True)
|
PREFIXES_DIR.mkdir(parents=True, exist_ok=True)
|
||||||
return PREFIXES_DIR / name
|
return PREFIXES_DIR / name
|
||||||
|
|
||||||
|
|
||||||
def safe_to_delete(path: Path) -> bool:
|
def safe_to_delete(path: Path) -> bool:
|
||||||
"""Only allow deleting paths inside PREFIXES_DIR or the default ~/.wine."""
|
"""Only allow deleting paths inside PREFIXES_DIR or the default ~/.wine.
|
||||||
default = Path.home() / ".wine"
|
|
||||||
|
Resolves symlinks/`..` first so a crafted name can't escape lexically."""
|
||||||
try:
|
try:
|
||||||
path.relative_to(PREFIXES_DIR)
|
resolved = path.resolve()
|
||||||
|
root = PREFIXES_DIR.resolve()
|
||||||
|
default = (Path.home() / ".wine").resolve()
|
||||||
|
except OSError:
|
||||||
|
return False
|
||||||
|
if resolved != root:
|
||||||
|
try:
|
||||||
|
resolved.relative_to(root)
|
||||||
return True
|
return True
|
||||||
except ValueError:
|
except ValueError:
|
||||||
pass
|
pass
|
||||||
return path == default
|
return resolved == default
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ taninux-Paket aus dem Repo heraus läuft.
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
|
import re
|
||||||
import sys
|
import sys
|
||||||
|
|
||||||
# ALLOWLIST: erste argv-Komponente → erlaubte Operationen.
|
# ALLOWLIST: erste argv-Komponente → erlaubte Operationen.
|
||||||
@@ -32,8 +33,9 @@ import sys
|
|||||||
# pacman/bootctl validieren die selbst, und sie können kein anderes Programm
|
# pacman/bootctl validieren die selbst, und sie können kein anderes Programm
|
||||||
# starten.
|
# starten.
|
||||||
#
|
#
|
||||||
# Wichtig: KEINE Shell, kein eval. Wir führen via os.execvp() exakt das
|
# Wichtig: KEINE Shell, kein eval. Wir führen via os.execv() exakt das
|
||||||
# übergebene argv aus, ohne Interpretation durch eine Shell.
|
# übergebene argv aus (Binary fest aus /usr/bin & co. aufgelöst, kein PATH),
|
||||||
|
# ohne Interpretation durch eine Shell.
|
||||||
ALLOWLIST: dict[str, dict[str, set[str]]] = {
|
ALLOWLIST: dict[str, dict[str, set[str]]] = {
|
||||||
# Pakete (de)installieren + Vollupgrade. --noconfirm erlaubt, damit das
|
# Pakete (de)installieren + Vollupgrade. --noconfirm erlaubt, damit das
|
||||||
# GUI-Frontend nicht in einem TTY-Prompt hängen bleibt.
|
# GUI-Frontend nicht in einem TTY-Prompt hängen bleibt.
|
||||||
@@ -127,6 +129,30 @@ class Rejected(Exception):
|
|||||||
"""Argv hat die Allowlist nicht passiert."""
|
"""Argv hat die Allowlist nicht passiert."""
|
||||||
|
|
||||||
|
|
||||||
|
# systemd-Unit-Namen enthalten nie Slashes — Pfad-Operanden (z.B. eine Unit-
|
||||||
|
# Datei ausserhalb der Unit-Suchpfade) werden abgelehnt.
|
||||||
|
_UNIT_NAME_RE = re.compile(r"^[A-Za-z0-9@._:\-]+$")
|
||||||
|
|
||||||
|
# Kommandos, deren Operanden Benutzer-/Gruppennamen (bzw. bei useradd -c ein
|
||||||
|
# Kommentar) sind — nie Pfade. Ohne den Check könnte z.B. `useradd -m ../x`
|
||||||
|
# ein Home-Verzeichnis an unerwarteter Stelle anlegen.
|
||||||
|
_NAME_OPERAND_CMDS = {"useradd", "userdel", "passwd", "gpasswd"}
|
||||||
|
|
||||||
|
# Erlaubte Verzeichnisse zum Auflösen des Kommandos — der Helper läuft als
|
||||||
|
# root, PATH darf die Wahl des Binaries nicht beeinflussen.
|
||||||
|
_SAFE_EXEC_DIRS = ("/usr/bin", "/usr/sbin", "/bin", "/sbin")
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_command(cmd: str) -> str | None:
|
||||||
|
"""Löst ein Allowlist-Kommando zu einem absoluten Pfad in den festen
|
||||||
|
System-Verzeichnissen auf (kein PATH-Lookup)."""
|
||||||
|
for d in _SAFE_EXEC_DIRS:
|
||||||
|
candidate = os.path.join(d, cmd)
|
||||||
|
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
|
||||||
|
return candidate
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def validate(argv: list[str]) -> list[str]:
|
def validate(argv: list[str]) -> list[str]:
|
||||||
"""Prüft argv gegen die ALLOWLIST. Gibt das (unveränderte) argv zurück
|
"""Prüft argv gegen die ALLOWLIST. Gibt das (unveränderte) argv zurück
|
||||||
oder wirft Rejected mit einer Begründung."""
|
oder wirft Rejected mit einer Begründung."""
|
||||||
@@ -152,7 +178,14 @@ def validate(argv: list[str]) -> list[str]:
|
|||||||
# ein Flag/Operator, der weder Operation noch erlaubtes Flag ist
|
# ein Flag/Operator, der weder Operation noch erlaubtes Flag ist
|
||||||
raise Rejected(f"option nicht erlaubt für {cmd}: {tok!r}")
|
raise Rejected(f"option nicht erlaubt für {cmd}: {tok!r}")
|
||||||
# ein Operand (Paketname, Boot-Eintrag) — keine Optionen-Injektion
|
# ein Operand (Paketname, Boot-Eintrag) — keine Optionen-Injektion
|
||||||
# möglich, da kein "-" am Anfang. Operanden sind frei.
|
# möglich, da kein "-" am Anfang.
|
||||||
|
if cmd == "systemctl":
|
||||||
|
if "/" in tok or ".." in tok or not _UNIT_NAME_RE.match(tok):
|
||||||
|
raise Rejected(f"ungültiger unit-name: {tok!r}")
|
||||||
|
elif cmd in _NAME_OPERAND_CMDS:
|
||||||
|
if "/" in tok or ".." in tok:
|
||||||
|
raise Rejected(f"ungültiger operand für {cmd}: {tok!r}")
|
||||||
|
# Alle anderen Operanden sind frei — pacman/bootctl/… validieren selbst.
|
||||||
|
|
||||||
# Kommandos mit erlaubten Operationen müssen eine davon nennen. Kommandos
|
# Kommandos mit erlaubten Operationen müssen eine davon nennen. Kommandos
|
||||||
# OHNE Ops (z.B. cupsenable/cupsdisable: nur Operand = Druckername) sind
|
# OHNE Ops (z.B. cupsenable/cupsdisable: nur Operand = Druckername) sind
|
||||||
@@ -185,11 +218,21 @@ def main(raw_argv: list[str]) -> int:
|
|||||||
print(f"taninux-helper: ok: {' '.join(argv)}")
|
print(f"taninux-helper: ok: {' '.join(argv)}")
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
# erlaubtes Kommando ausführen — execvp ersetzt den Helper-Prozess,
|
# erlaubtes Kommando ausführen — exec ersetzt den Helper-Prozess,
|
||||||
# damit pkexec den echten Exit-Code des Kommandos durchreicht.
|
# damit pkexec den echten Exit-Code des Kommandos durchreicht.
|
||||||
|
# Kein execvp/PATH-Lookup: das Binary wird ausschliesslich in festen
|
||||||
|
# System-Verzeichnissen aufgelöst.
|
||||||
|
exe = _resolve_command(argv[0])
|
||||||
|
if exe is None:
|
||||||
|
print(
|
||||||
|
f"taninux-helper: kommando {argv[0]!r} nicht gefunden in "
|
||||||
|
f"{':'.join(_SAFE_EXEC_DIRS)}",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return 127
|
||||||
try:
|
try:
|
||||||
os.execvp(argv[0], argv)
|
os.execv(exe, argv)
|
||||||
except OSError as err: # Programm nicht gefunden o.ä.
|
except OSError as err:
|
||||||
print(f"taninux-helper: exec fehlgeschlagen: {err}", file=sys.stderr)
|
print(f"taninux-helper: exec fehlgeschlagen: {err}", file=sys.stderr)
|
||||||
return 127
|
return 127
|
||||||
|
|
||||||
|
|||||||
@@ -57,11 +57,26 @@ def _installed_with_versions() -> dict[str, str]:
|
|||||||
return out
|
return out
|
||||||
|
|
||||||
|
|
||||||
|
# Variant-Suffixe, längste zuerst — damit "-rt-lts" vor "-rt"/"-lts" matcht.
|
||||||
|
_VARIANT_SUFFIXES = sorted(
|
||||||
|
(name.removeprefix("linux-") for name, _ in KERNEL_VARIANTS if name != "linux"),
|
||||||
|
key=len, reverse=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _running_variant(running: str) -> str:
|
||||||
|
"""Kernel-Paketname zur laufenden Release (`uname -r`).
|
||||||
|
|
||||||
|
Arch hängt den Variant-Suffix ans Ende der Release (z.B.
|
||||||
|
"6.9.7-zen1-1-zen", "6.6.30-rt30-1-rt-lts"); vanilla "linux" hat keinen."""
|
||||||
|
for suffix in _VARIANT_SUFFIXES:
|
||||||
|
if running.endswith(f"-{suffix}"):
|
||||||
|
return f"linux-{suffix}"
|
||||||
|
return "linux"
|
||||||
|
|
||||||
|
|
||||||
def _is_running(variant: str, running: str) -> bool:
|
def _is_running(variant: str, running: str) -> bool:
|
||||||
if variant == "linux":
|
return variant == _running_variant(running)
|
||||||
return not any(s in running for s in ("zen", "lts", "hardened", "rt"))
|
|
||||||
suffix = variant.removeprefix("linux-")
|
|
||||||
return suffix in running
|
|
||||||
|
|
||||||
|
|
||||||
def list_kernels() -> tuple[list[KernelInfo], bootloader.BootloaderInfo]:
|
def list_kernels() -> tuple[list[KernelInfo], bootloader.BootloaderInfo]:
|
||||||
|
|||||||
@@ -6,6 +6,7 @@ import re
|
|||||||
import subprocess
|
import subprocess
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from taninux.core.pacman import run_c_locale
|
||||||
from taninux.maintain.types import Category, Classification, Finding
|
from taninux.maintain.types import Category, Classification, Finding
|
||||||
|
|
||||||
|
|
||||||
@@ -35,10 +36,7 @@ class PacmanOrphansScanner:
|
|||||||
|
|
||||||
def scan(self) -> list[Finding]:
|
def scan(self) -> list[Finding]:
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
r = run_c_locale(["pacman", "-Qdtq"], timeout=10)
|
||||||
["pacman", "-Qdtq"],
|
|
||||||
capture_output=True, text=True, check=False, timeout=10,
|
|
||||||
)
|
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return []
|
return []
|
||||||
pkgs = [p.strip() for p in r.stdout.splitlines() if p.strip()]
|
pkgs = [p.strip() for p in r.stdout.splitlines() if p.strip()]
|
||||||
@@ -64,10 +62,9 @@ class PacmanOrphansScanner:
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def _installed_sizes(pkgs: list[str]) -> dict[str, int]:
|
def _installed_sizes(pkgs: list[str]) -> dict[str, int]:
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
# LC_ALL=C: das Parsing unten keyt auf die englischen Feldnamen
|
||||||
["pacman", "-Qi", *pkgs],
|
# "Name" / "Installed Size".
|
||||||
capture_output=True, text=True, check=False, timeout=30,
|
r = run_c_locale(["pacman", "-Qi", *pkgs], timeout=30)
|
||||||
)
|
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return {}
|
return {}
|
||||||
sizes: dict[str, int] = {}
|
sizes: dict[str, int] = {}
|
||||||
@@ -192,10 +189,7 @@ class PacmanCacheScanner:
|
|||||||
@staticmethod
|
@staticmethod
|
||||||
def _installed_packages() -> set[str]:
|
def _installed_packages() -> set[str]:
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
r = run_c_locale(["pacman", "-Qq"], timeout=10)
|
||||||
["pacman", "-Qq"],
|
|
||||||
capture_output=True, text=True, check=False, timeout=10,
|
|
||||||
)
|
|
||||||
return {l.strip() for l in r.stdout.splitlines() if l.strip()}
|
return {l.strip() for l in r.stdout.splitlines() if l.strip()}
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return set()
|
return set()
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import time
|
|||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
from taninux.core.pacman import run_c_locale
|
||||||
from taninux.maintain.scanners._util import dir_size
|
from taninux.maintain.scanners._util import dir_size
|
||||||
from taninux.maintain.types import Category, Classification, Finding
|
from taninux.maintain.types import Category, Classification, Finding
|
||||||
|
|
||||||
@@ -194,9 +195,12 @@ class SystemKernelsScanner:
|
|||||||
return []
|
return []
|
||||||
|
|
||||||
try:
|
try:
|
||||||
r = subprocess.run(
|
# LC_ALL=C: das "is owned by"-Parsing unten braucht englische
|
||||||
|
# pacman-Meldungen — sonst matcht nichts und ALLE Kernel-Module
|
||||||
|
# (inkl. laufender Kernel!) würden als SAFE eingestuft.
|
||||||
|
r = run_c_locale(
|
||||||
["pacman", "-Qo", *[str(d) for d in kernel_dirs]],
|
["pacman", "-Qo", *[str(d) for d in kernel_dirs]],
|
||||||
capture_output=True, text=True, check=False, timeout=30,
|
timeout=30,
|
||||||
)
|
)
|
||||||
except (FileNotFoundError, subprocess.TimeoutExpired):
|
except (FileNotFoundError, subprocess.TimeoutExpired):
|
||||||
return []
|
return []
|
||||||
@@ -209,6 +213,13 @@ class SystemKernelsScanner:
|
|||||||
if m:
|
if m:
|
||||||
owned.add(m.group(1).rstrip("/"))
|
owned.add(m.group(1).rstrip("/"))
|
||||||
|
|
||||||
|
# Fail closed: es gibt Kernel-Verzeichnisse, aber pacman meldet für
|
||||||
|
# KEINES einen Owner? Dann ist eher die Ownership-Abfrage kaputt
|
||||||
|
# (Locale, pacman-Fehler) als dass wirklich alles verwaist ist —
|
||||||
|
# nichts als löschbar einstufen.
|
||||||
|
if not owned:
|
||||||
|
return []
|
||||||
|
|
||||||
findings: list[Finding] = []
|
findings: list[Finding] = []
|
||||||
for d in kernel_dirs:
|
for d in kernel_dirs:
|
||||||
key = str(d).rstrip("/")
|
key = str(d).rstrip("/")
|
||||||
|
|||||||
@@ -1,6 +1,8 @@
|
|||||||
"""Update-Sources Registry."""
|
"""Update-Sources Registry."""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import sys
|
||||||
|
|
||||||
from taninux.update.sources import aur, flatpak, fwupd, pacman
|
from taninux.update.sources import aur, flatpak, fwupd, pacman
|
||||||
from taninux.update.types import UpdateItem
|
from taninux.update.types import UpdateItem
|
||||||
|
|
||||||
@@ -15,9 +17,11 @@ SOURCES = {
|
|||||||
|
|
||||||
def fetch_all() -> list[UpdateItem]:
|
def fetch_all() -> list[UpdateItem]:
|
||||||
items: list[UpdateItem] = []
|
items: list[UpdateItem] = []
|
||||||
for fn in SOURCES.values():
|
for name, fn in SOURCES.items():
|
||||||
try:
|
try:
|
||||||
items += fn()
|
items += fn()
|
||||||
except Exception: # noqa: BLE001
|
except Exception as e: # noqa: BLE001
|
||||||
pass
|
# Nicht stumm schlucken — sonst sieht eine kaputte Source wie
|
||||||
|
# "system up to date" aus. API bleibt stabil (Liste der Items).
|
||||||
|
print(f"taninux: update source {name} failed: {e}", file=sys.stderr)
|
||||||
return items
|
return items
|
||||||
|
|||||||
Reference in New Issue
Block a user