#!/usr/bin/env opsi-python # Generated by tools/build_single_file.py; edit the original sources, not this file. # Requires Python >= 3.13 and click; rich-click help is optional. Dependencies are not bundled. from __future__ import annotations from types import SimpleNamespace as _HelperNamespace def _bundled_helper(): """Secret-safe data operations for the opsi-secure wizard (standard library only).""" import argparse import copy import glob import grp import json import os import pwd import re import secrets import shlex import socket import stat import subprocess import sys import tempfile import time import tomllib from contextlib import contextmanager from pathlib import Path from typing import TYPE_CHECKING, Any from urllib.parse import parse_qs, quote, unquote, urlsplit if TYPE_CHECKING: from collections.abc import Generator from types import ModuleType PROTOCOL_IDS = ("clientconfig.depot.protocol", "clientconfig.depot.protocol.netboot") FEATURES = ("status-page", "rpc-interface", "messagebus_execute_process") REDIS_FILE = Path("/etc/redis/redis.conf") OPSI_FILE = Path("/etc/opsi/opsiconfd.conf") OPSI_GLOBAL_FILE = Path("/etc/opsi/opsi.conf") SOCKET = "/run/redis/opsi-secure.sock" REDIS_DROPIN = Path("/etc/systemd/system/redis-server.service.d/opsi-secure.conf") OPSI_DROPIN = Path("/etc/systemd/system/opsiconfd.service.d/opsi-secure-redis.conf") OPSI_DROPIN_TEXT = "[Service]\nSupplementaryGroups=redis\n" ACCOUNT_FILES = {name: Path("/etc") / name for name in ("passwd", "shadow", "group", "nsswitch.conf")} def language() -> str: """Resolve the system message locale dynamically, without changing process locale. The first nonempty LC_ALL, LC_MESSAGES or LANG wins. Only valid German locale names select German; LANGUAGE is deliberately not used. """ locale = next((os.environ[key] for key in ("LC_ALL", "LC_MESSAGES", "LANG") if os.environ.get(key)), "") return "de" if re.fullmatch(r"de(?:[_-][A-Za-z]{2})?(?:\.[A-Za-z0-9_-]+)?(?:@[A-Za-z0-9_-]+)?", locale) else "en" def msg(text: str, **values: object) -> str: """Translate an explicit message template before substituting non-secret values.""" template = text if language() == "de" else MESSAGES.get(text, text) return template.format_map(values) if values else template def output(text: str, *, file: Any = None) -> None: """Print a catalog message or already localized, explicitly selected data.""" print(msg(text), file=file) def framework_msg(text: str) -> str: """Translate framework-owned English templates, not user values or identifiers.""" return FRAMEWORK_MESSAGES.get(text, text) if language() == "de" else text @contextmanager def framework_translation(modules: list[ModuleType]) -> Generator[None]: """Scope framework gettext hooks to CLI rendering/parsing and restore on every exit. Click and argparse use module-local gettext aliases rather than an injectable translator. No process locale or environment is changed, and the original hooks are restored even on parser errors or interrupts. """ saved = [ (module, name, getattr(module, name)) for module in modules for name in ("_", "gettext", "ngettext") if hasattr(module, name) ] try: for module, name, _ in saved: setattr(module, name, framework_plural if name == "ngettext" else framework_msg) yield finally: for module, name, original in reversed(saved): setattr(module, name, original) def framework_plural(singular: str, plural: str, count: int) -> str: """Select the English/German plural rule without using the host gettext locale.""" return framework_msg(singular if count == 1 else plural) FRAMEWORK_MESSAGES: dict[str, str] = { "Secret-safe data operations for the opsi-secure wizard (standard library only).": "Geheimnissichere Datenoperationen für den opsi-secure-Assistenten (nur Standardbibliothek).", "Usage:": "Aufruf:", "Options": "Optionen", "Commands": "Befehle", "Arguments": "Argumente", "[OPTIONS]": "[OPTIONEN]", "[COMMAND] [ARGS]...": "[BEFEHL] [ARGUMENTE]...", "Error": "Fehler", "Show this message and exit.": "Diese Hilfe anzeigen und beenden.", "Show the version and exit.": "Version anzeigen und beenden.", "%(prog)s, version %(version)s": "%(prog)s, Version %(version)s", "Error: {message}": "Fehler: {message}", "Try '{command} {option}' for help.": "Für Hilfe '{command} {option}' aufrufen.", "No such option {name!r}.": "Unbekannte Option {name!r}.", "No such command {name!r}.": "Unbekannter Befehl {name!r}.", "Did you mean {possibility}?": "War {possibility} gemeint?", "(Did you mean one of: {possibilities}?)": "(War eine dieser Optionen gemeint: {possibilities}?)", "Missing command.": "Befehl fehlt.", "Missing argument": "Argument fehlt", "Missing option": "Option fehlt", "Missing parameter": "Parameter fehlt", "Missing parameter: {param_name}": "Parameter fehlt: {param_name}", "Missing {param_type}": "{param_type} fehlt", "Invalid value for {param_hint}: {message}": "Ungültiger Wert für {param_hint}: {message}", "Invalid value: {message}": "Ungültiger Wert: {message}", "Got unexpected extra argument ({args})": "Unerwartetes zusätzliches Argument ({args})", "Got unexpected extra arguments ({args})": "Unerwartete zusätzliche Argumente ({args})", "Option {name!r} does not take a value.": "Option {name!r} akzeptiert keinen Wert.", "Option {name!r} requires an argument.": "Option {name!r} benötigt ein Argument.", "Option {name!r} requires {nargs} arguments.": "Option {name!r} benötigt {nargs} Argumente.", "Argument {name!r} takes {nargs} values.": "Argument {name!r} benötigt {nargs} Werte.", "Aborted!": "Abgebrochen!", "Aborted.": "Abgebrochen.", "default: {default}": "Voreinstellung: {default}", "env var: {var}": "Umgebungsvariable: {var}", "file": "Datei", "directory": "Verzeichnis", "path": "Pfad", "{name} {filename!r} does not exist.": "{name} {filename!r} existiert nicht.", "{name} {filename!r} is a directory.": "{name} {filename!r} ist ein Verzeichnis.", "{name} {filename!r} is a file.": "{name} {filename!r} ist eine Datei.", "{name} {filename!r} is not readable.": "{name} {filename!r} ist nicht lesbar.", "{name} {filename!r} is not writable.": "{name} {filename!r} ist nicht schreibbar.", "{name} {filename!r} is not executable.": "{name} {filename!r} ist nicht ausführbar.", "{value!r} is not one of {choices}.": "{value!r} ist nicht einer der Werte {choices}.", "{value!r} is not {choice}.": "{value!r} ist nicht {choice}.", "usage: ": "Aufruf: ", "options": "Optionen", "positional arguments": "Positionsargumente", "show this help message and exit": "Diese Hilfe anzeigen und beenden", "%(prog)s: error: %(message)s\n": "%(prog)s: Fehler: %(message)s\n", "argument %(argument_name)s: %(message)s": "Argument %(argument_name)s: %(message)s", "invalid choice: %(value)r (choose from %(choices)s)": "Ungültige Auswahl: %(value)r (Auswahl: %(choices)s)", "the following arguments are required: %s": "Folgende Argumente sind erforderlich: %s", "unrecognized arguments: %s": "Unbekannte Argumente: %s", "ignored explicit argument %r": "Unerwartetes explizites Argument %r", "expected one argument": "Ein Argument erforderlich", "expected at least one argument": "Mindestens ein Argument erforderlich", "ambiguous option: %(option)s could match %(matches)s": "Mehrdeutige Option: %(option)s könnte %(matches)s entsprechen", } # German source messages remain stable identifiers. Unknown text (including # technical data and already translated messages) is passed through unchanged. MESSAGES: dict[str, str] = { "Ungeschütztes opsi-Backupverzeichnis.": "Unprotected opsi backup directory.", "opsi-Backupziel bereits vorhanden; keine Überschreibung.": "opsi backup target already exists; refusing to overwrite.", "opsi-Backupdatei nicht erfolgreich erstellt.": "opsi backup file was not created successfully.", "opsiconfd-Backup erstellt (root-only): {target}; Maßnahmen können beginnen.": "opsiconfd backup created (root-only): {target}; measures can begin.", "Werkzeug fehlt oder Zeitgrenze überschritten; Rohdaten unterdrückt.": "Tool missing or timeout exceeded; raw data suppressed.", "Externer Befehl fehlgeschlagen; Rohdaten unterdrückt.": "External command failed; raw data suppressed.", "Ungültige JSON-RPC-Antwort; Inhalt unterdrückt.": "Invalid JSON-RPC response; content suppressed.", "opsiconfd-Konfiguration konnte nicht validiert werden.": "Could not validate the opsiconfd configuration.", "Nicht reguläre Konfigurationsdatei; Änderung verweigert.": "Configuration file is not a regular file; change refused.", "Mehrfach definierte Zieloption; zuerst Konflikt auflösen.": "Target option defined more than once; resolve the conflict first.", "Nicht verifizierte Funktion.": "Unverified feature.", "API lieferte fremde Konfigurations-IDs; Abbruch.": "API returned unrelated configuration IDs; aborting.", "Unvollständiges ConfigState-Objekt; Abbruch.": "Incomplete ConfigState object; aborting.", "Beide Depot-Konfigurationsdefinitionen erforderlich.": "Both depot configuration definitions are required.", "WebDAV ist in den installierten Konfigurationsdefinitionen nicht unterstützt.": "WebDAV is not supported by the installed configuration definitions.", "Unbekannte systemd-Drop-ins; wirksame Konfiguration nicht eindeutig.": "Unknown systemd drop-ins; effective configuration is ambiguous.", "EnvironmentFile überschreibt möglicherweise Optionen; manuelle Prüfung erforderlich.": "EnvironmentFile may override options; manual verification required.", "Dienst-Umgebungsüberschreibungen erkannt; Änderung übersprungen.": "Service environment overrides detected; change skipped.", "Laufzeit-Umgebungsüberschreibungen erkannt; Änderung übersprungen.": "Runtime environment overrides detected; change skipped.", "Startparameter überschreiben Zieloptionen.": "Startup arguments override target options.", "Nicht unterstützter Redis-Dienststart.": "Unsupported Redis service startup.", "Abweichende opsiconfd-Konfigurationsdatei.": "Unexpected opsiconfd configuration file.", "opsiconfd-Includes nicht sicher auflösbar; unverändert lassen.": "Cannot safely resolve opsiconfd includes; leave unchanged.", "opsiconfd/JSON-RPC nach Neustart nicht verfügbar.": "opsiconfd/JSON-RPC unavailable after restart.", "Unsicheres Backup-Verzeichnis.": "Unsafe backup directory.", "Symlink im Backup-Ziel; Abbruch.": "Symlink in backup target; aborting.", "Nicht erlaubtes Wiederherstellungsziel.": "Restore target is not allowed.", "Symlink bei Wiederherstellung.": "Symlink encountered during restoration.", "Funktionen bereits deaktiviert; keine Änderung.": "Features already disabled; no change.", "Validierung der Funktionskonfiguration fehlgeschlagen.": "Feature configuration validation failed.", "Funktionskonfiguration nicht wirksam.": "Feature configuration is not effective.", "Funktionsänderung fehlgeschlagen; alte Datei und Dienst wiederhergestellt.": "Feature change failed; previous file and service restored.", "Funktionskonfiguration und JSON-RPC geprüft; keine Client-Funktionstests.": "Feature configuration and JSON-RPC verified; no client functionality tests.", "Effektive Depotprotokolle nicht eindeutig lesbar.": "Effective depot protocols cannot be read unambiguously.", "Nicht alle Depots besitzen eine WebDAV-URL.": "Not all depots have a WebDAV URL.", "Depot bereits im Zielzustand; keine Änderung.": "Depot already in the target state; no change.", "Depotänderung nicht vollständig wirksam.": "Depot change is not fully effective.", "Effektive Depotprotokolle weichen ab.": "Effective depot protocols differ.", "API-Rollback nicht vollständig verifiziert; Backup manuell wiederherstellen.": "API rollback not fully verified; restore the backup manually.", "API-Änderung und Rollback fehlgeschlagen; geschützte Objektsicherungen verwenden.": "API change and rollback failed; use the protected object backups.", "API-Änderung fehlgeschlagen; Depotobjekte wiederhergestellt.": "API change failed; depot objects restored.", "Depotdefinitionen, genaue Overrides und effektive API-Werte geprüft; Client/Netboot nicht getestet.": "Depot definitions, exact overrides and effective API values verified; client/netboot not tested.", "Zyklische Redis-Includes.": "Cyclic Redis includes.", "Redis-Syntax nicht eindeutig analysierbar.": "Redis syntax cannot be parsed unambiguously.", "Relative Redis-Includes nicht eindeutig auflösbar.": "Relative Redis includes cannot be resolved unambiguously.", "Unvollständige Redis-Antwort.": "Incomplete Redis response.", "Ungültiger Redis-Bulk-Header.": "Invalid Redis bulk header.", "Redis-Bulk-Antwort außerhalb der Größenbegrenzung.": "Redis bulk response exceeds the size limit.", "Unvollständige Redis-Bulk-Antwort.": "Incomplete Redis bulk response.", "Nicht der vorgesehene Redis-Socket.": "Not the intended Redis socket.", "Redis-Authentifizierung fehlt.": "Redis authentication is missing.", "Geheimnistragende opsiconfd-Konfiguration nicht auf root/Dienstbenutzer und Administratorgruppe beschränkt.": "Secret-bearing opsiconfd configuration is not restricted to root/service user and administrator group.", "Socket-Rechte nicht minimal.": "Socket permissions are not minimal.", "Socket-Eigentümer/Gruppe weichen ab.": "Socket owner/group differ.", "Authentifizierter Redis-Zugriff fehlgeschlagen.": "Authenticated Redis access failed.", "Authentifizierter Redis-PING fehlgeschlagen.": "Authenticated Redis PING failed.", "Redis verweigert nicht authentifizierten Zugriff nicht.": "Redis does not refuse unauthenticated access.", "Redis besitzt noch TCP/TLS-Listener.": "Redis still has TCP/TLS listeners.", "opsiconfd/JSON-RPC nicht erreichbar.": "opsiconfd/JSON-RPC is unreachable.", "opsiconfd-Socket-Verbindung nicht nachweisbar.": "Cannot verify an opsiconfd socket connection.", "Migration nur bei zuvor aktiven Kerndiensten; Dienstzustand bleibt unverändert.": "Migration requires previously active core services; service state remains unchanged.", "Socket, Authentifizierung, API und fehlende TCP-Listener geprüft.": "Socket, authentication, API and absence of TCP listeners verified.", "Externe oder abweichende Redis-Installation; Migration übersprungen.": "External or nonstandard Redis installation; migration skipped.", "Bestehende Redis-Authentifizierung/ACL wird nicht ungefragt ersetzt oder rotiert.": "Existing Redis authentication/ACL will not be replaced or rotated without approval.", "Redis-Includes wurden analysiert, automatische Migration bleibt bei Includes gesperrt.": "Redis includes analyzed; automatic migration remains blocked when includes are present.", "Replikation/Cluster ist kein dedizierter lokaler Redis.": "Replication/cluster is not a dedicated local Redis.", "Abweichender opsiconfd-Dienstbenutzer.": "Unexpected opsiconfd service user.", "Abweichende Redis-Dienstidentität.": "Unexpected Redis service identity.", "Abweichende Redis-Ports/TLS-Installation; manuell migrieren.": "Nonstandard Redis ports/TLS installation; migrate manually.", "Weiterer oder nicht identifizierbarer Redis-Verbraucher verbunden.": "Additional or unidentified Redis consumer connected.", "Nicht lokaler Redis-Verbraucher.": "Nonlocal Redis consumer.", "Weiterer Redis-Konfigurationsverbraucher; Migration übersprungen.": "Additional Redis configuration consumer; migration skipped.", "Lokaler Redis; verbundene Verbraucher opsiconfd. Inaktive/externe Verbraucher nicht vollständig ausschließbar: Betreiberbestätigung erforderlich.": "Local Redis; connected consumer opsiconfd. Inactive/external consumers cannot be fully ruled out: operator confirmation required.", "Redis bereits abgesichert; kein Passwortwechsel und kein Neustart.": "Redis already secured; no password change or restart.", "opsiconfd akzeptiert Redis-URL nicht.": "opsiconfd does not accept the Redis URL.", "Redis-Testkonfiguration startet nicht; Produktionsdienste unverändert.": "Redis test configuration does not start; production services unchanged.", "Redis-Testkonfiguration nicht bereit.": "Redis test configuration is not ready.", "Redis-Testauthentifizierung fehlgeschlagen.": "Redis test authentication failed.", "Redis-Migration und Rollback fehlgeschlagen; SSH/Root-Wiederherstellung erforderlich.": "Redis migration and rollback failed; SSH/root recovery required.", "Redis-Migration fehlgeschlagen; bisherige Dateien und Dienste wiederhergestellt.": "Redis migration failed; previous files and services restored.", "Redis: authentifiziertes Socket-PING, NOAUTH, keine TCP/TLS-Listener, opsiconfd und JSON-RPC geprüft.": "Redis: authenticated socket PING, NOAUTH, absence of TCP/TLS listeners, opsiconfd and JSON-RPC verified.", "Ungültiger oder geschützter Depotbenutzer; keine Kontolöschung.": "Invalid or protected depot user; no account deletion.", "Globale opsi-Konfiguration nicht lesbar oder ungültig; keine Kontolöschung.": "Global opsi configuration unreadable or invalid; no account deletion.", "Depotbenutzer fehlt in der globalen opsi-Konfiguration; keine Kontolöschung.": "Depot user missing from global opsi configuration; no account deletion.", "Konfigurierter Depotbenutzer geändert; keine Kontolöschung.": "Configured depot user changed; no account deletion.", "laufender Prozess {pid}": "running process {pid}", "Depotbenutzer nur nach WebDAV-/lokaler Konten-/Abhängigkeitsprüfung und Zustimmung (interaktiv oder --yes/-y) entfernen. Bestandsaufnahme löscht keine Konten.": "Remove the depot user only after WebDAV/local account/dependency checks and consent (interactive or --yes/-y). Inventory does not delete accounts.", "Depotbenutzer bleibt erhalten: lokaler Kontozustand nicht eindeutig.": "Depot user retained: local account state is ambiguous.", "Depotbenutzer ist kein lokales Benutzerkonto (z. B. Domänenkonto); keine Löschung.": "Depot user is not a local user account (e.g. a domain account); no deletion.", "Depotbenutzer bleibt erhalten: Depotprotokolle nicht vollständig WebDAV.": "Depot user retained: depot protocols are not fully WebDAV.", "Nur ein eindeutig lokales, unprivilegiertes Depotkonto darf entfernt werden.": "Only an unambiguously local, unprivileged depot account may be removed.", "Ungültige lokale Depotkonto-Identität; keine Löschung.": "Invalid local depot account identity; no deletion.", "Depotbenutzer bleibt erhalten: NSS-Identität weicht vom lokalen Konto ab.": "Depot user retained: NSS identity differs from the local account.", "Lokale Kontendaten nicht eindeutig; keine Löschung.": "Local account data is ambiguous; no deletion.", "Geteilte UID; Depotbenutzer bleibt erhalten.": "Shared UID; depot user retained.", "Depotbenutzer besitzt Dateien; Dateibesitz verhindert die Kontolöschung.": "Depot user owns files; file ownership prevents account deletion.", "Depotbenutzer hat laufende oder konfigurierte Abhängigkeiten; keine Kontolöschung.": "Depot user has running or configured dependencies; no account deletion.", "Konfigurierte Abhängigkeit vom Depotbenutzer; Konto bleibt erhalten.": "Configured dependency on depot user; account retained.", "Systemabhängigkeit vom Depotbenutzer; Konto bleibt erhalten.": "System dependency on depot user; account retained.", "Mount-Inventar nicht lesbar oder ungültig; keine Kontolöschung.": "Mount inventory unreadable or invalid; no account deletion.", "Automount-Dateisystem nicht vollständig prüfbar; keine Kontolöschung.": "Automount filesystem cannot be fully checked; no account deletion.", "Besitzprüfung auf zusätzlichem Daten-Dateisystem fehlgeschlagen; keine Kontolöschung.": "Ownership check on additional data filesystem failed; no account deletion.", "Dateibesitz auf einem weiteren Dateisystem verhindert die Kontolöschung.": "File ownership on another filesystem prevents account deletion.", "Ungültige Kontosicherung.": "Invalid account backup.", "Unsichere Kontosicherung.": "Unsafe account backup.", "Ungeschützte Kontosicherung.": "Unprotected account backup.", "Kontosicherung nicht lesbar oder ungültig.": "Account backup unreadable or invalid.", "Gruppen-ID inzwischen geändert; Wiederherstellung verweigert.": "Group ID has changed; restoration refused.", "Gesicherte UID inzwischen vergeben.": "Saved UID is now assigned to another account.", "Depotbenutzer inzwischen verändert; Wiederherstellung verweigert.": "Depot user has changed; restoration refused.", "Lokaler Kontozustand nicht eindeutig; keine Kontolöschung.": "Local account state is ambiguous; no account deletion.", "Depotbenutzer ist kein eindeutig lokales Benutzerkonto; keine Kontolöschung.": "Depot user is not an unambiguously local user account; no account deletion.", "Nur ein unprivilegiertes Depotkonto darf entfernt werden.": "Only an unprivileged depot account may be removed.", "Ungültige lokale Depotkonto-Identität; keine Kontolöschung.": "Invalid local depot account identity; no account deletion.", "NSS-Identität weicht vom lokalen Konto ab; keine Kontolöschung.": "NSS identity differs from the local account; no account deletion.", "Depotbenutzer bereits entfernt; keine Änderung.": "Depot user already removed; no change.", "Depotkonto während Prüfung geändert; keine Kontolöschung.": "Depot account changed during verification; no account deletion.", "Lokale Gruppendaten nicht eindeutig; keine Kontolöschung.": "Local group data is ambiguous; no account deletion.", "Primäre Gruppe nicht eindeutig lokal vorhanden; keine Kontolöschung.": "Primary group is not unambiguously present locally; no account deletion.", "Kontozustand während Prüfung geändert; keine Kontolöschung.": "Account state changed during verification; no account deletion.", "Depotbenutzer-Löschung nicht wirksam.": "Depot user deletion is not effective.", "Gruppenidentität inzwischen verändert.": "Group identity has changed.", "Kontolöschung oder Gruppenprüfung fehlgeschlagen; kein automatischer Konto-Rollback. Keine separate Kontosicherung; manueller Abgleich von Konto und Gruppen erforderlich.": "Account deletion or group verification failed; no automatic account rollback. No separate account backup; manual account and group verification required.", "Depotbenutzer entfernt; Home, Depotdaten und Gruppen erhalten. Keine Kontosicherung oder automatische Wiederherstellung.": "Depot user removed; home, depot data and groups preserved. No account backup or automatic restoration.", "Nicht die von diesem Werkzeug eingerichtete Redis-Konfiguration.": "Not the Redis configuration created by this tool.", "Bestehendes Redis-Passwort nicht verfügbar; unverändert.": "Existing Redis password unavailable; unchanged.", "URI-Reparatur fehlgeschlagen; ursprüngliche Konfiguration wiederhergestellt.": "URI repair failed; original configuration restored.", "URI-Logredaktion korrigiert, bestehendes Passwort erhalten; Socket/API geprüft.": "URI log redaction corrected, existing password preserved; socket/API verified.", "Unmaskiertes Redis-Passwort in Dienstlogs gefunden; sichere Bereinigung und ausdrückliche Rotation erforderlich.": "Unmasked Redis password found in service logs; secure cleanup and explicit rotation required.", "Dienstlogs und Journal geprüft: kein unmaskiertes Redis-Passwort gefunden (nur Wahrheitswert, keine Logzeilen ausgegeben).": "Service logs and journal checked: no unmasked Redis password found (boolean only, no log lines displayed).", "Administrative API: ": "Administrative API: ", "verfügbar": "available", "nicht bestätigt": "not confirmed", "Deaktivierte Funktionen: ": "Disabled features: ", "keine": "none", "Lizenzen: ": "Licenses: ", "fehlt": "missing", "nicht gesetzt": "not set", "leer": "empty", "Explizite Overrides: {count}": "Explicit overrides: {count}", "Depots mit WebDAV-URL: {ready}/{count}": "Depots with WebDAV URL: {ready}/{count}", "Produkte: {products}; offene Aktionen: {actions}": "Products: {products}; pending actions: {actions}", "bereits abgesichert": "already secured", "noch nicht abgesichert": "not yet secured", "vorhanden; Bestandsaufnahme ohne Löschung": "present; inventory without deletion", "nicht vorhanden": "not present", "Unbekannte Bestandsanzeige.": "Unknown inventory display.", "opsiconfd/JSON-RPC nicht verfügbar.": "opsiconfd/JSON-RPC unavailable.", "opsiconfd und frische administrative JSON-RPC-Anmeldung funktionieren.": "opsiconfd and fresh administrative JSON-RPC authentication work.", "Redis-Socket, Authentifizierung, NOAUTH und fehlende TCP/TLS-Listener geprüft.": "Redis socket, authentication, NOAUTH and absence of TCP/TLS listeners verified.", "Prüfung fehlgeschlagen; sensible Fehlerdetails unterdrückt.": "Verification failed; sensitive error details suppressed.", "Dienstzustand nicht lesbar; Rohdaten unterdrückt.": "Service state unreadable; raw data suppressed.", "Ungültige Samba-Sicherung.": "Invalid Samba backup.", "Nicht unterstützter Samba-Dienstzustand.": "Unsupported Samba service state.", "Doppelte Samba-Einheit in Sicherung.": "Duplicate Samba unit in backup.", " [automatisch: Ja]": " [automatic: Yes]", " [j/N]: ": " [y/N]: ", "WARNUNG: Automatische Zustimmung aktiv: ALLE Fragen werden mit Ja beantwortet, einschließlich initialer Sicherung, WebDAV-/SMB-Risiken, Override-Löschung, Dienständerungen, Funktionsabschaltungen, Kontolöschung des konfigurierten Depotbenutzers und Wiederherstellung. Kontolöschung ohne separate Kontosicherung oder automatischen Konto-Rollback; Wiederherstellung kann neuere Änderungen überschreiben. Schutzprüfungen bleiben verpflichtend.": "WARNING: Automatic consent active: ALL questions will be answered Yes, including initial backup, WebDAV/SMB risks, override deletion, service changes, feature disabling, account deletion of the configured depot user and restoration. Account deletion has no separate account backup or automatic account rollback; restoration may overwrite newer changes. Safety checks remain mandatory.", "RISIKO: Capture/Clonezilla können Schreibzugriff verlieren; externe SMB-Nutzung und Automatisierung können ausfallen. Auch inaktive SMB-Abhängigkeiten sind nicht sicher ausschließbar.": "RISK: Capture/Clonezilla may lose write access; external SMB usage and automation may fail. Even inactive SMB dependencies cannot be safely ruled out.", "sollen jetzt beide Depot-Standardwerte ({ids}) auf WebDAV umgestellt werden": "should both depot defaults ({ids}) now be changed to WebDAV", "sollen explizite Overrides beider Konfigurationen gesichert und gelöscht werden (Depot und/oder Client Werte; {ids})": "should explicit overrides of both configurations be backed up and deleted (depot and/or client values; {ids})", "soll jetzt die Samba-Deaktivierung geprüft werden": "should Samba disabling now be checked", "Sind Client-/Netboot-Komponenten und relevante Produkte für WebDAV geprüft, werden diese Risiken akzeptiert und {action}?": "Have client/netboot components and relevant products been checked for WebDAV, are these risks accepted and {action}?", "{measure}: Kompatibilität geprüft und SMB-Risiko akzeptiert": "{measure}: compatibility checked and SMB risk accepted", "Backup im Dry-Run verweigert.": "Backup refused in dry-run.", "Geschützte Sicherung: {backup}": "Protected backup: {backup}", "PLAN: Vor den Maßnahmen ein opsiconfd-Backup in einem geschützten Verzeichnis erstellen.": "PLAN: Create an opsiconfd backup in a protected directory before the measures.", "=== Initiale Sicherung ===": "=== Initial backup ===", "Das opsiconfd-Backup enthält die opsi-Konfiguration, das Backend und Redis-Daten.": "The opsiconfd backup contains the opsi configuration, backend and Redis data.", "Linux-Konten aus passwd/shadow werden durch das opsiconfd-Backup nicht gesichert.": "Linux accounts from passwd/shadow are not backed up by the opsiconfd backup.", "Eine kurze Wartungsunterbrechung ist möglich; Ablage in einem geschützten, nur für root zugänglichen Verzeichnis.": "A brief maintenance interruption is possible; stored in a protected directory accessible only to root.", "Vor den Maßnahmen ein opsiconfd-Backup erstellen? Der Server wird dafür in den Maintenance Mode (Wartungsmodus) versetzt.": "Create an opsiconfd backup before the measures? The server will enter Maintenance Mode for this.", "Keine Zustimmung zur initialen Sicherung; keine Maßnahmen ausgeführt.": "No consent to the initial backup; no measures performed.", "Änderungsvorbereitung im Dry-Run verweigert.": "Change preparation refused in dry-run.", "Initiale Sicherung fehlgeschlagen.": "Initial backup failed.", "Root-Rechte erforderlich; keine Änderung.": "Root privileges required; no change.", "Nur Debian 13 freigegeben.": "Only Debian 13 is approved.", "Erforderliches Werkzeug fehlt: {tool}": "Required tool missing: {tool}", "Andere opsi-secure-Ausführung aktiv.": "Another opsi-secure execution is active.", "Paketbasierte opsi-4.3-/Redis-/Bootimage-Installation auf Debian 13 geprüft.": "Package-based opsi 4.3/Redis/bootimage installation on Debian 13 verified.", "=== Bestandsaufnahme (keine Konfigurationsänderung) ===": "=== Inventory (no configuration change) ===", "Bestandsaufnahme fehlgeschlagen.": "Inventory failed.", "Depotprotokolle, Overrides und Produkte": "Depot protocols, overrides and products", "Konto adminuser": "Account adminuser", "=== Konto {name} ===": "=== Account {name} ===", "Depotbenutzer-Bestandsaufnahme fehlgeschlagen; sensible Fehlerdetails unterdrückt.": "Depot user inventory failed; sensitive error details suppressed.", "=== Samba-Einheiten und Freigaben ===": "=== Samba units and shares ===", "Samba-Einheiten: ": "Samba units: ", "WARNUNG: Samba-Einheiten konnten nicht gelesen werden; Samba-Deaktivierung gesperrt.": "WARNING: Could not read Samba units; Samba disabling blocked.", "testparm nicht verfügbar.": "testparm unavailable.", "Samba-Konfiguration lesbar; Freigaben: {shares} (Details mit --verbose).": "Samba configuration readable; shares: {shares} (details with --verbose).", "WARNUNG: Samba-Freigaben konnten mit testparm nicht gelesen werden; Samba-Deaktivierung gesperrt.": "WARNING: Could not read Samba shares with testparm; Samba disabling blocked.", "WebDAV-Standardwerte bereits gesetzt; keine expliziten Overrides, keine Änderung.": "WebDAV defaults already set; no explicit overrides, no change.", "PLAN: Beide WebDAV-Standardwerte nach gemeinsamer Kompatibilitäts-/Risikobestätigung umstellen.": "PLAN: Change both WebDAV defaults after joint compatibility/risk confirmation.", "PLAN: Explizite Overrides nur nach separater Zustimmung sichern und löschen.": "PLAN: Back up and delete explicit overrides only after separate consent.", "Beide Standardwerte auf webdav; alle übrigen Definitionseigenschaften bleiben erhalten.": "Both defaults to webdav; all other definition properties preserved.", "WebDAV unverändert: keine Zustimmung.": "WebDAV unchanged: no consent.", "Explizite Overrides beider Konfigurationen sichern und löschen (Depot und/oder Client Werte). Betroffene Konfigurationen: {ids}?": "Back up and delete explicit overrides of both configurations (depot and/or client values). Affected configurations: {ids}?", "Explizite Overrides bleiben erhalten; vollständige WebDAV-Umstellung nicht bestätigt.": "Explicit overrides retained; full WebDAV migration not confirmed.", "Bereits abgesichert und validiert; keine Sicherung, kein Passwortwechsel, kein Neustart.": "Already secured and validated; no backup, password change or restart.", "Lokaler Socket mit Passwort, TCP/TLS aus. Kein Passwort wird ausgegeben.": "Local socket with password, TCP/TLS off. No password is displayed.", "Kurze Unterbrechung: opsiconfd stoppen, Redis neu starten, opsiconfd starten.": "Brief interruption: stop opsiconfd, restart Redis, start opsiconfd.", "Includes, bestehende ACL/Passwörter, weitere Verbraucher und Overrides sperren die Migration.": "Includes, existing ACL/passwords, additional consumers and overrides block migration.", "Auch inaktive/periodische Verbraucher müssen geprüft sein; Socket 0660, RuntimeDirectory 0750.": "Inactive/periodic consumers must also be checked; socket 0660, RuntimeDirectory 0750.", "PLAN: 256-Bit-Passwort, Socket 0660, RuntimeDirectory 0750, Dienstgruppe redis, TCP/TLS aus.": "PLAN: 256-bit password, socket 0660, RuntimeDirectory 0750, service group redis, TCP/TLS off.", "Wird Redis ausschließlich für dieses lokale opsi verwendet und soll jetzt abgesichert werden?": "Is Redis used exclusively for this local opsi and should it now be secured?", "=== Administrator, 2FA und Aussperrungsschutz ===": "=== Administrator, 2FA and lockout protection ===", "Ohne opsi_auth-Lizenz ist die automatische Migration auf einen alternativen Administrator nicht verfügbar.": "Without an opsi_auth license, automatic migration to an alternative administrator is unavailable.", "Ohne 2fa-Lizenz ist kein 2FA für den Administrator möglich.": "Without a 2fa license, 2FA is not available for the administrator.", "Administratoranlage/2FA unsupported: Auch mit Lizenzen ist der Loginmechanismus hier nicht validiert.": "Administrator creation/2FA unsupported: even with licenses, the login mechanism is not validated here.", "Bestehende PAM-Administration behalten; unterstützten Datenbank-Workflow beim Hersteller klären.": "Retain existing PAM administration; clarify the supported database workflow with the vendor.", "Kein Secret erzeugt, kein TOTP zurückgesetzt, kein lokales adminuser-Konto gelöscht.": "No secret generated, no TOTP reset, no local adminuser account deleted.", "Linux-Konto, User-Objekt, Passwort-Hash, Gruppenberechtigung und TOTP sind getrennte Mechanismen.": "Linux account, User object, password hash, group permissions and TOTP are separate mechanisms.", "user_setCredentials betrifft NUR den Depotbenutzer, nicht einen neuen Administrator!": "user_setCredentials affects ONLY the depot user, not a new administrator!", "opsiconfd 4.3.59.1 benötigt opsi_auth für Datenbank-Anmeldung und 2fa für TOTP-Registrierung.": "opsiconfd 4.3.59.1 requires opsi_auth for database login and 2fa for TOTP registration.", "totp_mandatory betrifft Passwort-Benutzer einschließlich Nichtadmins; Hostauth/Tokens haben Ausnahmen.": "totp_mandatory affects password users including nonadmins; host authentication/tokens have exceptions.", "=== Nicht benötigte Funktionen ===": "=== Unneeded features ===", "Ungültige Funktionskonfiguration.": "Invalid feature configuration.", "Vorhandene disabled-features: {features}": "Existing disabled-features: {features}", "Nicht authentifizierte /status-Seite aus; Status-Monitoring kann betroffen sein.": "Unauthenticated /status page off; status monitoring may be affected.", "Nur RPC-Werkzeug auf der Admin-Seite aus. Benötigte JSON-RPC-API /rpc bleibt an.": "Only the RPC tool on the admin page is disabled. The required JSON-RPC API /rpc remains enabled.", "Server-Prozessausführung über Messagebus aus; Remote-Prozessworkflows können ausfallen.": "Server process execution via messagebus off; remote process workflows may fail.", "{feature}: Bereits deaktiviert; keine Änderung.": "{feature}: already disabled; no change.", "PLAN: {feature} nach separater Zustimmung ergänzen.": "PLAN: Add {feature} after separate consent.", "Funktion {feature} deaktivieren? opsiconfd-Neustart am Ende dieser Auswahl erforderlich.": "Disable feature {feature}? An opsiconfd restart is required at the end of this selection.", "ÜBERSPRUNGEN: Samba-Bestandsaufnahme unvollständig; Samba-Deaktivierung gesperrt.": "SKIPPED: Samba inventory incomplete; Samba disabling blocked.", "Nicht unterstützter Dienstzustand; unverändert.": "Unsupported service state; unchanged.", "Samba bereits deaktiviert oder keine Samba-Units vorhanden; keine Änderung.": "Samba already disabled or no Samba units present; no change.", "PLAN: Samba nur deaktivieren, wenn BEIDE Depotprotokolle samt Overrides effektiv WebDAV sind.": "PLAN: Disable Samba only if BOTH depot protocols including overrides are effectively WebDAV.", " Separate Zustimmung zur Deaktivierung; Zustand sichern, nur ausstehende Units stoppen/deaktivieren.": " Separate consent to disabling; back up state, stop/disable only pending units.", "ÜBERSPRUNGEN: Samba bleibt aktiv. Beide Depotprotokolle müssen vollständig auf WebDAV stehen.": "SKIPPED: Samba remains active. Both depot protocols must be fully WebDAV.", "Defaults, explizite Overrides und effektive Client-/Netboot-Werte werden gemeinsam geprüft.": "Defaults, explicit overrides and effective client/netboot values are checked together.", "OK: Beide Depotprotokolle, Overrides und effektive Werte sind WebDAV.": "OK: Both depot protocols, overrides and effective values are WebDAV.", "Samba-Dienste und deren Aktivierungseinheiten stoppen/deaktivieren (keine Pakete oder Nutzdaten löschen)?": "Stop/disable Samba services and their activation units (do not delete packages or user data)?", "ÜBERSPRUNGEN: Depotprotokolle inzwischen geändert; Samba bleibt unverändert.": "SKIPPED: Depot protocols have changed; Samba remains unchanged.", "Samba-Deaktivierung nicht wirksam.": "Samba disabling is not effective.", "Samba-Änderung fehlgeschlagen; gespeicherte Zustände wiederhergestellt.": "Samba change failed; saved states restored.", "Samba-Einheiten gestoppt/deaktiviert; Freigaben und Nutzdaten erhalten. Kein SMB-Funktionstest.": "Samba units stopped/disabled; shares and user data preserved. No SMB functionality test.", "=== Benutzer ===": "=== Users ===", "Lokales adminuser-Konto wird nicht gelöscht; bestehende PAM-Administration bleibt erhalten.": "Local adminuser account will not be deleted; existing PAM administration retained.", "Auch mit Lizenzen bleibt die Administrator-/2FA-Migration hier unsupported: Loginmechanismus nicht validiert.": "Even with licenses, administrator/2FA migration remains unsupported here: login mechanism not validated.", "Konfigurierter Depotbenutzer: {name}": "Configured depot user: {name}", "Depotbenutzer bleibt erhalten: Konfiguration, WebDAV oder Abhängigkeitsprüfung nicht freigegeben.": "Depot user retained: configuration, WebDAV or dependency check not approved.", "{name} bereits entfernt; keine Änderung.": "{name} already removed; no change.", "{name}: Nach vollständiger WebDAV-Umstellung kann das lokale Konto entfernt werden. Home/Depotdaten und Gruppen bleiben erhalten.": "{name}: After full WebDAV migration, the local account can be removed. Home/depot data and groups preserved.", "WARNUNG: Keine separate Kontosicherung und keine automatische Wiederherstellung des gelöschten Linux-Benutzers. Das opsiconfd-Backup sichert keine Linux-Konten aus passwd/shadow.": "WARNING: No separate account backup and no automatic restoration of the deleted Linux user. The opsiconfd backup does not back up Linux accounts from passwd/shadow.", "PLAN: {name} nach WebDAV-/Abhängigkeitsprüfung und separater Zustimmung entfernen (ohne Kontosicherung).": "PLAN: Remove {name} after WebDAV/dependency checks and separate consent (without account backup).", "Lokales Konto {name} jetzt entfernen?": "Remove local account {name} now?", "Symlink in Wiederherstellung verweigert.": "Symlink refused during restoration.", "Mehrdeutige Depotkonto-Sicherungen; keine Wiederherstellung.": "Ambiguous depot account backups; no restoration.", "Ungeschützte Depotkonto-Sicherung.": "Unprotected depot account backup.", "Geschützten Backup-Zustand wiederherstellen? Neuere Änderungen können verloren gehen; betroffene Dienste separat prüfen.": "Restore the protected backup state? Newer changes may be lost; check affected services separately.", "Dateien wiederhergestellt; Dienste manuell prüfen/starten. Keine API-Prüfung oder Dienstneustarts.": "Files restored; check/start services manually. No API verification or service restarts.", "Depot-Wiederherstellung fehlgeschlagen.": "Depot restoration failed.", "Wiederherstellungsprüfung fehlgeschlagen.": "Restoration verification failed.", "Wiederherstellung geprüft soweit unterstützt; keine Transaktionsgarantie für API/Dateien/Dienste/Konten.": "Restoration verified where supported; no transaction guarantee for API/files/services/accounts.", "=== Plan und Anwendung ===": "=== Plan and application ===", "Maßnahme fehlgeschlagen; Rohfehler/Geheimnisse werden nicht protokolliert.": "Measure failed; raw errors/secrets are not logged.", "=== Abschlussprüfung ===": "=== Final verification ===", "Konfigurations-/lokale Integrationstests sind keine realen Client-/Netboot-/2FA-Funktionstests.": "Configuration/local integration tests are not real client/netboot/2FA functionality tests.", "Bewusst akzeptierte Funktionsverluste:": "Explicitly accepted functionality losses:", "Backups und Wiederherstellung: {backup} (root-only), --restore mit separater Bestätigung.": "Backups and restoration: {backup} (root-only), --restore with separate confirmation.", "Ergebniscode: {result} (30 = teilweise übersprungen, nicht vollständig gehärtet).": "Result code: {result} (30 = partially skipped, not fully hardened).", "=== Prüfung (keine Konfigurationsänderung) ===": "=== Verification (no configuration change) ===", "Dry-Run mit Wiederherstellung verweigert; keine Änderung.": "Dry-run with restoration refused; no change.", "Anwendung nur in einem direkten interaktiven Terminal oder mit explizitem --yes/-y.": "Application requires a direct interactive terminal or explicit --yes/-y.", "Abbruch: Maßnahme fehlgeschlagen; Rohfehler/Geheimnisse werden nicht protokolliert.": "Aborted: measure failed; raw errors/secrets are not logged.", "Abbruch: Aufräumen fehlgeschlagen; Rohfehler unterdrückt.": "Aborted: cleanup failed; raw errors suppressed.", "ALLE Fragen mit Ja beantworten, einschließlich Backup, Risiken, destruktiver Kontolöschung, Dienständerungen und Wiederherstellung; ohne direktes Terminal. Bei check/Dry-Run inaktiv.": "Answer ALL questions Yes, including backup, risks, destructive account deletion, service changes and restoration; without a direct terminal. Inactive for check/dry-run.", "Exit-Codes: 0 geprüft/ausgeführt, 2 Argumente/Voraussetzungen, 10 parallele Ausführung, 20 fehlgeschlagen, 30 teilweise übersprungen.": "Exit codes: 0 verified/performed, 2 arguments/prerequisites, 10 concurrent execution, 20 failed, 30 partially skipped.", "Keine Backups, Geheimnisse, Änderungen oder Neustarts.": "No backups, secrets, changes or restarts.", "Gefilterte JSON-Details und zusätzliche Erläuterungen anzeigen.": "Show filtered JSON details and additional explanations.", "Gefilterte JSON-Details anzeigen.": "Show filtered JSON details.", "Nur prüfen und Maßnahmen planen, ohne Bestätigungen.": "Only check and plan measures, without confirmations.", "Zusätzliche Erläuterungen anzeigen.": "Show additional explanations.", "Bestandsaufnahme und Validierung ohne Änderungen oder Bestätigungen.": "Inventory and validation without changes or confirmations.", "Maßnahmen einzeln bestätigen, anwenden und anschließend validieren.": "Confirm measures individually, apply and then validate.", "Geschützten Lauf wiederherstellen, einschließlich API und Dienstneustarts.": "Restore a protected run, including API and service restarts.", "Interaktive opsi-4.3-Härtung auf Debian 13.\n\nRoot und paketbasierte Installation erforderlich. Ohne --yes/-y Bestätigungen\nnur am direkten Terminal, Voreinstellung Nein. Lesende API-Aufrufe können serverseitig\nSitzungsdaten und Zugriffslogs erzeugen. Authentifizierungs-/2FA-Migration und\nadminuser-Löschung bleiben unsupported. Konfigurierten Depotbenutzer nur nach Schutzprüfung entfernen.": "Interactive opsi 4.3 hardening on Debian 13.\n\nRoot and a package-based installation required. Without --yes/-y, confirmations\nrequire a direct terminal and default to No. Read-only API calls may create server-side\nsession data and access logs. Authentication/2FA migration and\nadminuser deletion remain unsupported. Remove the configured depot user only after safety checks.", "Geschütztes Maßnahmenbackup nur als Dateien wiederherstellen.\n\nAPI-unabhängig; Root und Verzeichnislock erforderlich; ohne --yes/-y direktes Terminal.\nKein opsi-cli nötig. Keine Dienstneustarts; Dienste danach manuell prüfen/starten.": "Restore a protected measure backup as files only.\n\nAPI-independent; root and directory lock required; direct terminal without --yes/-y.\nNo opsi-cli needed. No service restarts; check/start services manually afterwards.", "Python-Laufzeitabhängigkeit fehlt: click erforderlich; für diesen Interpreter installieren.": "Python runtime dependency missing: click required; install it for this interpreter.", "Abbruch: CLI fehlgeschlagen; Rohfehler unterdrückt.": "Aborted: CLI failed; raw errors suppressed.", "Python-CLI-Abhängigkeit fehlt; optionale rich-click-/rich-Installation prüfen.": "Python CLI dependency missing; check the optional rich-click/rich installation.", "Abbruch; Rohfehler unterdrückt.": "Aborted; raw errors suppressed.", } class Refusal(Exception): """An operation is unsupported, unsafe or failed; messages contain no raw external data.""" def __init__(self, text: str) -> None: """Localize only the explicitly supplied, secret-safe refusal message.""" super().__init__(msg(text)) class AccountRefusal(Refusal): """A statically worded account precondition refusal safe to display to the operator.""" def opsi_backup(directory: Path) -> None: """Create a protected opsiconfd baseline before hardening, without forwarding output.""" info = directory.lstat() if ( directory.is_symlink() or not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700 ): raise Refusal("Ungeschütztes opsi-Backupverzeichnis.") target = directory / "opsi-backup.msgpack.lz4" if target.exists() or target.is_symlink(): raise Refusal("opsi-Backupziel bereits vorhanden; keine Überschreibung.") previous = os.umask(0o077) try: run(["opsiconfd", "backup", "--quiet", "--log-level-stderr", "0", str(target)], timeout=600) info = target.lstat() if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_size == 0: raise Refusal("opsi-Backupdatei nicht erfolgreich erstellt.") target.chmod(0o600) finally: os.umask(previous) output(msg("opsiconfd-Backup erstellt (root-only): {target}; Maßnahmen können beginnen.", target=target)) def subprocess_environment() -> dict[str, str]: """Isolate system tools from frozen Python libraries and environmental credentials.""" env = dict(os.environ, SYSTEMD_PAGER="cat", PAGER="cat", SYSTEMD_COLORS="0") if getattr(sys, "frozen", False): # PyInstaller prepends its private libraries; system tools must use their own. original = env.pop("LD_LIBRARY_PATH_ORIG", None) env.pop("LD_LIBRARY_PATH", None) if original: env["LD_LIBRARY_PATH"] = original for name in list(env): if name.startswith(("OPSI_CLI_PASSWORD", "OPSI_CLI_TOTP")) or name == "REDISCLI_AUTH": env.pop(name) return env def run(args: list[str], *, data: str | None = None, timeout: int = 45) -> str: """Run a bounded command without forwarding errors or environmental credentials.""" env = subprocess_environment() try: result = subprocess.run(args, input=data, text=True, capture_output=True, timeout=timeout, env=env, check=False) except (OSError, subprocess.TimeoutExpired): raise Refusal("Werkzeug fehlt oder Zeitgrenze überschritten; Rohdaten unterdrückt.") from None if result.returncode: raise Refusal("Externer Befehl fehlgeschlagen; Rohdaten unterdrückt.") return result.stdout def api(method: str, *params: object) -> Any: """Pass JSON-RPC's final parameter via stdin, never via process arguments.""" args = [ "opsi-cli", "--log-level-file", "0", "--log-level-stderr", "0", "--output-format", "json", "--non-interactive", ] data = None if params: args += ["--input-file", "-"] data = json.dumps(params[-1]) args += ["jsonrpc", "execute", method] args += [json.dumps(param) for param in params[:-1]] try: return json.loads(run(args, data=data, timeout=60)) except (ValueError, TypeError): raise Refusal("Ungültige JSON-RPC-Antwort; Inhalt unterdrückt.") from None def effective_config(path: Path = OPSI_FILE) -> dict[str, Any]: """Parse opsiconfd's own configuration output in memory, without displaying secrets.""" try: return json.loads(run(["opsiconfd", "-c", str(path), "get-config"])) except (ValueError, TypeError): raise Refusal("opsiconfd-Konfiguration konnte nicht validiert werden.") from None def atomic_write( path: Path, text: str, *, mode: int | None = None, uid: int | None = None, gid: int | None = None ) -> None: """Replace a regular file atomically, retaining metadata unless explicitly hardened.""" if path.is_symlink() or (path.exists() and not path.is_file()): raise Refusal("Nicht reguläre Konfigurationsdatei; Änderung verweigert.") old = path.stat() if path.exists() else None fd, name = tempfile.mkstemp(prefix=".opsi-secure-", dir=path.parent) try: with os.fdopen(fd, "w") as stream: os.fchmod(stream.fileno(), mode if mode is not None else stat.S_IMODE(old.st_mode) if old else 0o600) if os.geteuid() == 0: os.fchown( stream.fileno(), uid if uid is not None else old.st_uid if old else 0, gid if gid is not None else old.st_gid if old else 0, ) stream.write(text) stream.flush() os.fsync(stream.fileno()) os.replace(name, path) dir_fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(dir_fd) finally: os.close(dir_fd) finally: if os.path.exists(name): os.unlink(name) def edit_config(text: str, updates: dict[str, str], *, redis: bool = False) -> str: """Keep comments/unknown lines; refuse ambiguous duplicates and replace selected values.""" lines = text.splitlines(keepends=True) separator = " " if redis else " = " for key, value in updates.items(): pattern = re.compile(r"^(\s*)" + re.escape(key) + (r"\s+([^#\n]*)(.*)$" if redis else r"\s*=\s*([^#\n]*)(.*)$")) matches = [(idx, pattern.match(line.rstrip("\n"))) for idx, line in enumerate(lines)] matches = [(idx, match) for idx, match in matches if match] if len(matches) > 1: raise Refusal("Mehrfach definierte Zieloption; zuerst Konflikt auflösen.") if matches: idx, match = matches[0] comment = (" " + match.group(3).lstrip()) if match.group(3).strip() else "" lines[idx] = match.group(1) + key + separator + value + comment + "\n" else: if lines and not lines[-1].endswith("\n"): lines[-1] += "\n" lines.append(key + separator + value + "\n") return "".join(lines) def merge_features(existing: list[str], additions: list[str]) -> list[str]: """Preserve the order of existing entries and append supported additions once.""" if any(feature not in FEATURES for feature in additions): raise Refusal("Nicht verifizierte Funktion.") return list(dict.fromkeys([*existing, *additions])) def protocol_objects(objects: list[dict[str, Any]], *, states: bool = False) -> list[dict[str, Any]]: """Reject broad or malformed API responses before backing up/updating/deleting anything.""" key = "configId" if states else "id" if any(obj.get(key) not in PROTOCOL_IDS for obj in objects): raise Refusal("API lieferte fremde Konfigurations-IDs; Abbruch.") if states and any(not obj.get("objectId") or not isinstance(obj.get("values"), list) for obj in objects): raise Refusal("Unvollständiges ConfigState-Objekt; Abbruch.") if not states: if len(objects) != 2 or {obj["id"] for obj in objects} != set(PROTOCOL_IDS): raise Refusal("Beide Depot-Konfigurationsdefinitionen erforderlich.") if any("webdav" not in obj.get("possibleValues", []) for obj in objects): raise Refusal("WebDAV ist in den installierten Konfigurationsdefinitionen nicht unterstützt.") return copy.deepcopy(objects) def webdav_defaults(objects: list[dict[str, Any]]) -> list[dict[str, Any]]: """Change only defaultValues on the two fully preserved config objects.""" result = protocol_objects(objects) for obj in result: obj["defaultValues"] = ["webdav"] return result def unit_property(unit: str, prop: str) -> str: """Read a systemd property without a pager or raw error forwarding.""" return run(["systemctl", "show", "--value", "-p", prop, unit]).strip() def overrides_clear(*, redis: bool = False) -> None: """Reject service/runtime overrides that could shadow edits; only our exact drop-ins are accepted.""" units = ["opsiconfd.service"] + (["redis-server.service"] if redis else []) for unit in units: allowed = {str(OPSI_DROPIN)} if unit == "opsiconfd.service" else {str(REDIS_DROPIN)} if set(unit_property(unit, "DropInPaths").split()) - allowed: raise Refusal("Unbekannte systemd-Drop-ins; wirksame Konfiguration nicht eindeutig.") if unit_property(unit, "EnvironmentFiles"): raise Refusal("EnvironmentFile überschreibt möglicherweise Optionen; manuelle Prüfung erforderlich.") environment = unit_property(unit, "Environment") if re.search(r"OPSICONFD_|(?:^|\s)REDIS_", environment): raise Refusal("Dienst-Umgebungsüberschreibungen erkannt; Änderung übersprungen.") pid = unit_property(unit, "MainPID") if pid.isdigit() and int(pid): entries = Path(f"/proc/{pid}/environ").read_bytes().split(b"\0") if any(entry.startswith((b"OPSICONFD_", b"REDIS_")) for entry in entries): raise Refusal("Laufzeit-Umgebungsüberschreibungen erkannt; Änderung übersprungen.") cmdline = ( Path(f"/proc/{pid}/cmdline").read_bytes().decode().replace("\0", " ") if pid.isdigit() and int(pid) else "" ) if unit == "opsiconfd.service" and re.search( r"--(?:config-file|redis|disabled-features|multi-factor|run-as)|(?:^|\s)-c\s", cmdline ): raise Refusal("Startparameter überschreiben Zieloptionen.") if unit == "redis-server.service": # Redis rewrites /proc/PID/cmdline to its listening address after startup. # Validate systemd's actual ExecStart, not that rewritten process title. start = unit_property(unit, "ExecStart") if "/etc/redis/redis.conf" not in start or re.search(r"--(?:port|tls|requirepass|unixsocket|acl)", start): raise Refusal("Nicht unterstützter Redis-Dienststart.") if effective_config().get("config_file") != str(OPSI_FILE): raise Refusal("Abweichende opsiconfd-Konfigurationsdatei.") if re.search(r"^\s*(?:include|config-file)\s*[= ]", OPSI_FILE.read_text(), re.MULTILINE): raise Refusal("opsiconfd-Includes nicht sicher auflösbar; unverändert lassen.") def service_ready() -> bool: """Require a running daemon and a fresh successful JSON-RPC call.""" try: run(["systemctl", "is-active", "--quiet", "opsiconfd"]) return api("accessControl_userIsAdmin") is True except Refusal: return False def wait_ready() -> None: """Retry actual API readiness, not just service startup, within a bounded window.""" for _ in range(12): if service_ready(): return time.sleep(2) raise Refusal("opsiconfd/JSON-RPC nach Neustart nicht verfügbar.") def backup_file(path: Path, directory: Path) -> None: """Save exact bytes and original metadata to a root-only backup directory.""" directory.mkdir(mode=0o700, parents=True, exist_ok=True) if directory.stat().st_uid != os.geteuid() or stat.S_IMODE(directory.stat().st_mode) != 0o700: raise Refusal("Unsicheres Backup-Verzeichnis.") key = str(path).lstrip("/").replace("/", "__") metadata: dict[str, Any] = {"path": str(path), "exists": path.exists()} if path.is_symlink(): raise Refusal("Symlink im Backup-Ziel; Abbruch.") if path.exists(): info = path.stat() metadata.update(mode=stat.S_IMODE(info.st_mode), uid=info.st_uid, gid=info.st_gid) atomic_write(directory / key, path.read_text(), mode=0o600) atomic_write(directory / (key + ".meta.json"), json.dumps(metadata), mode=0o600) def restore_files(directory: Path) -> None: """Restore only this program's allowlisted files, including original ownership and mode.""" allowed = {str(OPSI_FILE), str(REDIS_FILE), str(OPSI_DROPIN), str(REDIS_DROPIN)} for meta_path in directory.glob("*.meta.json"): metadata = json.loads(meta_path.read_text()) path = Path(metadata["path"]) if str(path) not in allowed: raise Refusal("Nicht erlaubtes Wiederherstellungsziel.") if metadata["exists"]: atomic_write( path, meta_path.with_name(meta_path.name.removesuffix(".meta.json")).read_text(), mode=metadata["mode"], uid=metadata["uid"], gid=metadata["gid"], ) elif path.exists(): if path.is_symlink(): raise Refusal("Symlink bei Wiederherstellung.") path.unlink() membership = directory / "redis-membership.json" if membership.exists() and json.loads(membership.read_text()).get("added"): account = pwd.getpwnam("opsiconfd") if grp.getgrnam("redis").gr_gid in os.getgrouplist("opsiconfd", account.pw_gid): run(["gpasswd", "-d", "opsiconfd", "redis"]) def feature_apply(additions: list[str], directory: Path) -> None: """Validate staged config, back up, commit once and roll back a failed daemon/API check.""" overrides_clear() config = effective_config() old_features = config.get("disabled_features", []) new_features = merge_features(old_features, additions) if old_features == new_features: output("Funktionen bereits deaktiviert; keine Änderung.") return text = edit_config(OPSI_FILE.read_text(), {"disabled-features": json.dumps(new_features)}) fd, name = tempfile.mkstemp(prefix=".opsi-secure-validate-", dir=OPSI_FILE.parent) os.close(fd) temp_path = Path(name) try: atomic_write(temp_path, text, mode=0o600) if effective_config(temp_path).get("disabled_features") != new_features: raise Refusal("Validierung der Funktionskonfiguration fehlgeschlagen.") finally: temp_path.unlink(missing_ok=True) backup_file(OPSI_FILE, directory) try: atomic_write(OPSI_FILE, text) run(["systemctl", "restart", "opsiconfd"]) wait_ready() if effective_config().get("disabled_features") != new_features: raise Refusal("Funktionskonfiguration nicht wirksam.") except (Refusal, OSError, ValueError, KeyError, TypeError, KeyboardInterrupt): restore_files(directory) run(["systemctl", "restart", "opsiconfd"]) wait_ready() raise Refusal("Funktionsänderung fehlgeschlagen; alte Datei und Dienst wiederhergestellt.") from None output("Funktionskonfiguration und JSON-RPC geprüft; keine Client-Funktionstests.") def depot_inventory() -> dict[str, Any]: """Read only protocol definitions/overrides and non-secret product/depot attributes.""" configs = protocol_objects(api("config_getObjects", [], {"id": list(PROTOCOL_IDS)})) states = protocol_objects(api("configState_getObjects", [], {"configId": list(PROTOCOL_IDS)}), states=True) products = api("product_getObjects", ["id", "type", "productVersion", "packageVersion"], {}) depots = api("host_getObjects", ["id", "depotWebdavUrl"], {"type": ["OpsiConfigserver", "OpsiDepotserver"]}) actions = api("productOnClient_getObjects", ["productId", "clientId", "actionRequest", "installationStatus"], {}) return { "configs": configs, "states": states, "products": products, "depots": [{"id": obj["id"], "webdav": bool(obj.get("depotWebdavUrl"))} for obj in depots], "active_actions": [ {k: obj.get(k) for k in ("productId", "clientId", "actionRequest", "installationStatus")} for obj in actions if obj.get("actionRequest") not in (None, "none") ], "effective": api("configState_getValues", list(PROTOCOL_IDS), [], True), } def depot_webdav_ready() -> bool: """Require WebDAV for both defaults, every override and every effective client value.""" state = depot_inventory() configs = protocol_objects(state["configs"]) overrides = protocol_objects(state["states"], states=True) if any(obj.get("defaultValues") != ["webdav"] for obj in configs): return False if any(obj["values"] != ["webdav"] for obj in overrides): return False effective = state["effective"] if not isinstance(effective, dict): raise Refusal("Effektive Depotprotokolle nicht eindeutig lesbar.") return all( isinstance(values, dict) and all(values.get(key) == ["webdav"] for key in PROTOCOL_IDS) for values in effective.values() ) def depot_apply(directory: Path, *, delete_overrides: bool) -> None: """Update exact configs and optionally exact states, restoring both on failure.""" before = depot_inventory() if not all(obj["webdav"] for obj in before["depots"]): raise Refusal("Nicht alle Depots besitzen eine WebDAV-URL.") new = webdav_defaults(before["configs"]) states = before["states"] if delete_overrides else [] if new == before["configs"] and not states: output("Depot bereits im Zielzustand; keine Änderung.") return directory.mkdir(mode=0o700, parents=True, exist_ok=True) atomic_write(directory / "configs.json", json.dumps(before["configs"]), mode=0o600) atomic_write(directory / "states.json", json.dumps(states), mode=0o600) try: if new != before["configs"]: api("config_updateObjects", new) if states: api("configState_deleteObjects", protocol_objects(states, states=True)) after = depot_inventory() if after["configs"] != new or (delete_overrides and after["states"]): raise Refusal("Depotänderung nicht vollständig wirksam.") if delete_overrides and any( values.get(key) != ["webdav"] for values in after["effective"].values() for key in PROTOCOL_IDS ): raise Refusal("Effektive Depotprotokolle weichen ab.") except (Refusal, OSError, ValueError, KeyError, TypeError, KeyboardInterrupt): try: api("config_updateObjects", before["configs"]) if states: api("configState_createObjects", states) check = depot_inventory() if check["configs"] != before["configs"] or check["states"] != before["states"]: raise Refusal("API-Rollback nicht vollständig verifiziert; Backup manuell wiederherstellen.") except Refusal: raise Refusal("API-Änderung und Rollback fehlgeschlagen; geschützte Objektsicherungen verwenden.") from None raise Refusal("API-Änderung fehlgeschlagen; Depotobjekte wiederhergestellt.") from None output("Depotdefinitionen, genaue Overrides und effektive API-Werte geprüft; Client/Netboot nicht getestet.") def redis_lines(path: Path, seen: set[Path] | None = None) -> list[list[str]]: """Resolve effective Redis include order for discovery; do not expose directive values.""" seen = set() if seen is None else seen resolved = path.resolve() if resolved in seen: raise Refusal("Zyklische Redis-Includes.") seen.add(resolved) result: list[list[str]] = [] for line in path.read_text().splitlines(): try: parts = shlex.split(line, comments=True) except ValueError: raise Refusal("Redis-Syntax nicht eindeutig analysierbar.") from None if not parts: continue result.append(parts) if parts[0] == "include": for pattern in parts[1:]: if not pattern.startswith("/"): raise Refusal("Relative Redis-Includes nicht eindeutig auflösbar.") for filename in sorted(glob.glob(pattern)): result.extend(redis_lines(Path(filename), seen.copy())) return result def redis_exchange(sock: socket.socket, *parts: str) -> bytes: """Send a Redis RESP request, without process arguments or diagnostics containing passwords.""" encoded = [part.encode() for part in parts] request = b"*" + str(len(encoded)).encode() + b"\r\n" for part in encoded: request += b"$" + str(len(part)).encode() + b"\r\n" + part + b"\r\n" sock.sendall(request) result = b"" while not result.endswith(b"\r\n"): chunk = sock.recv(1) if not chunk or len(result) > 1024: raise Refusal("Unvollständige Redis-Antwort.") result += chunk if result.startswith(b"$"): try: size = int(result[1:-2]) except ValueError: raise Refusal("Ungültiger Redis-Bulk-Header.") from None if not 0 <= size <= 1024 * 1024: raise Refusal("Redis-Bulk-Antwort außerhalb der Größenbegrenzung.") remaining = size + 2 while remaining: chunk = sock.recv(min(remaining, 65536)) if not chunk: raise Refusal("Unvollständige Redis-Bulk-Antwort.") remaining -= len(chunk) result += chunk return result def redis_validate(config: dict[str, Any] | None = None) -> None: """Check socket permissions, authenticated/unauthenticated access and all Redis TCP listeners.""" config = effective_config() if config is None else config url = urlsplit(config.get("redis_internal_url", "")) if url.scheme != "unix" or url.path != SOCKET: raise Refusal("Nicht der vorgesehene Redis-Socket.") password = parse_qs(url.query).get("password", [None])[0] or (unquote(url.password) if url.password else None) if not password: raise Refusal("Redis-Authentifizierung fehlt.") file_info = OPSI_FILE.stat() if ( file_info.st_uid not in (0, pwd.getpwnam("opsiconfd").pw_uid) or file_info.st_gid != grp.getgrnam("opsiadmin").gr_gid or stat.S_IMODE(file_info.st_mode) not in (0o600, 0o640, 0o660) ): raise Refusal( "Geheimnistragende opsiconfd-Konfiguration nicht auf root/Dienstbenutzer und Administratorgruppe beschränkt." ) info = Path(SOCKET).stat() if not stat.S_ISSOCK(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o660: raise Refusal("Socket-Rechte nicht minimal.") if info.st_uid != pwd.getpwnam("redis").pw_uid or info.st_gid != grp.getgrnam("redis").gr_gid: raise Refusal("Socket-Eigentümer/Gruppe weichen ab.") for authenticated in (False, True): with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(5) sock.connect(SOCKET) if authenticated and redis_exchange(sock, "AUTH", password) != b"+OK\r\n": raise Refusal("Authentifizierter Redis-Zugriff fehlgeschlagen.") response = redis_exchange(sock, "PING") if authenticated and response != b"+PONG\r\n": raise Refusal("Authentifizierter Redis-PING fehlgeschlagen.") if not authenticated and not response.startswith(b"-NOAUTH"): raise Refusal("Redis verweigert nicht authentifizierten Zugriff nicht.") if "redis-server" in run(["ss", "-lntpH"]): raise Refusal("Redis besitzt noch TCP/TLS-Listener.") if not service_ready(): raise Refusal("opsiconfd/JSON-RPC nicht erreichbar.") # API auth/session creation requires Redis; additionally require live UNIX connections owned by opsiconfd. connections = run(["ss", "-xnpH"]) if "opsiconfd" not in connections or SOCKET not in connections: raise Refusal("opsiconfd-Socket-Verbindung nicht nachweisbar.") def redis_inventory() -> dict[str, Any]: """Determine migration eligibility conservatively; do not rotate pre-existing credentials.""" overrides_clear(redis=True) if any(unit_property(unit, "ActiveState") != "active" for unit in ("opsiconfd", "redis-server")): raise Refusal("Migration nur bei zuvor aktiven Kerndiensten; Dienstzustand bleibt unverändert.") config = effective_config() url = urlsplit(config.get("redis_internal_url", "")) lines = redis_lines(REDIS_FILE) settings = {parts[0]: parts[1:] for parts in lines} if url.scheme == "unix" and url.path == SOCKET: redis_validate(config) return { "eligible": True, "already_secure": True, "reason": msg("Socket, Authentifizierung, API und fehlende TCP-Listener geprüft."), } if url.scheme != "redis" or url.hostname not in ("localhost", "127.0.0.1", "::1"): raise Refusal("Externe oder abweichende Redis-Installation; Migration übersprungen.") if url.password or "requirepass" in settings or "user" in settings or "aclfile" in settings: raise Refusal("Bestehende Redis-Authentifizierung/ACL wird nicht ungefragt ersetzt oder rotiert.") if any(parts[0] == "include" for parts in lines): raise Refusal("Redis-Includes wurden analysiert, automatische Migration bleibt bei Includes gesperrt.") if settings.get("replicaof") or settings.get("slaveof") or settings.get("cluster-enabled") == ["yes"]: raise Refusal("Replikation/Cluster ist kein dedizierter lokaler Redis.") if config.get("run_as_user") != "opsiconfd": raise Refusal("Abweichender opsiconfd-Dienstbenutzer.") if unit_property("redis-server", "User") != "redis" or unit_property("redis-server", "Group") != "redis": raise Refusal("Abweichende Redis-Dienstidentität.") if settings.get("port", ["6379"]) != ["6379"] or int(settings.get("tls-port", ["0"])[0]): raise Refusal("Abweichende Redis-Ports/TLS-Installation; manuell migrieren.") # Current network connections are evidence, not proof of absence of inactive consumers. connections = run(["ss", "-ntpH"]) for line in connections.splitlines(): if ":6379" not in line: continue if "opsiconfd" not in line and "redis-server" not in line: raise Refusal("Weiterer oder nicht identifizierbarer Redis-Verbraucher verbunden.") with socket.create_connection((url.hostname, url.port or 6379), timeout=5) as sock: response = redis_exchange(sock, "CLIENT", "LIST") for line in response.decode(errors="replace").splitlines(): address = re.search(r"(?:^|\s)addr=([^ ]+)", line) if address and not address.group(1).startswith(("127.0.0.1:", "[::1]:", "::1:")): raise Refusal("Nicht lokaler Redis-Verbraucher.") # Reject active references elsewhere; comments in stock Grafana config are not consumers. for root in (Path("/etc"),): for path in root.rglob("*"): if ( not path.is_file() or path.is_symlink() or str(path).startswith(("/etc/redis/", "/etc/opsi/opsiconfd.conf")) ): continue if path.stat().st_size > 2_000_000: continue try: text = path.read_text() active = "\n".join(line for line in text.splitlines() if not line.lstrip().startswith(("#", ";"))) if re.search(r"redis[s]?://|(?:localhost|127\.0\.0\.1|::1):6379", active): raise Refusal("Weiterer Redis-Konfigurationsverbraucher; Migration übersprungen.") except (UnicodeError, PermissionError): continue return { "eligible": True, "already_secure": False, "reason": msg( "Lokaler Redis; verbundene Verbraucher opsiconfd. Inaktive/externe Verbraucher nicht vollständig ausschließbar: Betreiberbestätigung erforderlich." ), } def redis_apply(directory: Path) -> None: """Migrate socket+authentication as one reversible operation for the verified Debian service layout.""" inventory = redis_inventory() if inventory["already_secure"]: output("Redis bereits abgesichert; kein Passwortwechsel und kein Neustart.") return password = secrets.token_hex(32) redis_text = edit_config( REDIS_FILE.read_text(), {"port": "0", "tls-port": "0", "unixsocket": SOCKET, "unixsocketperm": "660", "requirepass": password}, redis=True, ) # opsiconfd registers url.password for its secret filter, but not query passwords. url = redis_connection_url(password) opsi_text = edit_config(OPSI_FILE.read_text(), {"redis-internal-url": url}) for path in (REDIS_FILE, OPSI_FILE, REDIS_DROPIN, OPSI_DROPIN): backup_file(path, directory) account = pwd.getpwnam("opsiconfd") add_group = grp.getgrnam("redis").gr_gid not in os.getgrouplist("opsiconfd", account.pw_gid) atomic_write(directory / "redis-membership.json", json.dumps({"added": add_group}), mode=0o600) atomic_write( directory / "services.json", json.dumps({unit: unit_property(unit, "ActiveState") for unit in ("opsiconfd", "redis-server")}), mode=0o600, ) # Validate opsiconfd parse BEFORE commit; never emit get-config's output. fd, filename = tempfile.mkstemp(prefix=".opsi-secure-redis-", dir=OPSI_FILE.parent) os.close(fd) temp = Path(filename) try: atomic_write(temp, opsi_text, mode=0o600) if effective_config(temp).get("redis_internal_url") != url: raise Refusal("opsiconfd akzeptiert Redis-URL nicht.") finally: temp.unlink(missing_ok=True) # Redis has no --test-config; validate staged config in a private temporary runtime directory. with tempfile.TemporaryDirectory(prefix="opsi-secure-redis-") as tmp: test_dir = Path(tmp) test_config = edit_config( redis_text, { "daemonize": "no", "supervised": "no", "logfile": '""', "dir": tmp, "pidfile": tmp + "/pid", "unixsocket": tmp + "/socket", "save": '""', "appendonly": "no", }, redis=True, ) atomic_write(test_dir / "redis.conf", test_config, mode=0o600) process = subprocess.Popen( ["redis-server", str(test_dir / "redis.conf")], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=subprocess_environment(), ) try: for _ in range(30): if process.poll() is not None: raise Refusal("Redis-Testkonfiguration startet nicht; Produktionsdienste unverändert.") if (test_dir / "socket").exists(): break time.sleep(0.1) else: raise Refusal("Redis-Testkonfiguration nicht bereit.") with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: sock.settimeout(3) sock.connect(str(test_dir / "socket")) if redis_exchange(sock, "AUTH", password) != b"+OK\r\n" or redis_exchange(sock, "PING") != b"+PONG\r\n": raise Refusal("Redis-Testauthentifizierung fehlgeschlagen.") finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) try: run(["systemctl", "stop", "opsiconfd"]) # opsiconfd rebuilds supplementary groups via os.setgroups(getgrouplist()). # A systemd SupplementaryGroups= entry alone is therefore not sufficient. if add_group: run(["usermod", "-aG", "redis", "opsiconfd"]) atomic_write(REDIS_FILE, redis_text, mode=0o640, uid=0, gid=grp.getgrnam("redis").gr_gid) atomic_write(OPSI_FILE, opsi_text, mode=0o640, uid=0, gid=grp.getgrnam("opsiadmin").gr_gid) REDIS_DROPIN.parent.mkdir(mode=0o755, parents=True, exist_ok=True) OPSI_DROPIN.parent.mkdir(mode=0o755, parents=True, exist_ok=True) atomic_write(REDIS_DROPIN, "[Service]\nRuntimeDirectoryMode=0750\n", mode=0o644) atomic_write(OPSI_DROPIN, OPSI_DROPIN_TEXT, mode=0o644) run(["systemctl", "daemon-reload"]) run(["systemctl", "restart", "redis-server"]) run(["systemctl", "start", "opsiconfd"]) wait_ready() redis_validate() except (Refusal, OSError, ValueError, KeyError, TypeError, KeyboardInterrupt): try: restore_files(directory) run(["systemctl", "daemon-reload"]) run(["systemctl", "restart", "redis-server"]) run(["systemctl", "restart", "opsiconfd"]) wait_ready() except Refusal: raise Refusal( "Redis-Migration und Rollback fehlgeschlagen; SSH/Root-Wiederherstellung erforderlich." ) from None raise Refusal("Redis-Migration fehlgeschlagen; bisherige Dateien und Dienste wiederhergestellt.") from None output("Redis: authentifiziertes Socket-PING, NOAUTH, keine TCP/TLS-Listener, opsiconfd und JSON-RPC geprüft.") def _validate_depot_user(name: object) -> str: """Reject unsafe names and known privileged or service-account targets.""" if ( not isinstance(name, str) or len(name) > 32 or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_-]*\$?", name) is None or name in { "root", "adminuser", "opsiconfd", "redis", "opsi", "daemon", "bin", "sys", "sync", "games", "man", "lp", "mail", "news", "uucp", "proxy", "www-data", "backup", "list", "irc", "_apt", "nobody", "systemd-network", "systemd-timesync", "messagebus", "sshd", } ): raise AccountRefusal("Ungültiger oder geschützter Depotbenutzer; keine Kontolöschung.") return name def configured_depot_user() -> str: """Read the validated depot username strictly from the global opsi TOML configuration.""" try: config = tomllib.loads(OPSI_GLOBAL_FILE.read_text(encoding="utf-8")) except (OSError, UnicodeError, tomllib.TOMLDecodeError): raise AccountRefusal("Globale opsi-Konfiguration nicht lesbar oder ungültig; keine Kontolöschung.") from None section = config.get("depot_user") if not isinstance(section, dict) or "username" not in section: raise AccountRefusal("Depotbenutzer fehlt in der globalen opsi-Konfiguration; keine Kontolöschung.") return _validate_depot_user(section["username"]) def _require_configured_depot_user(name: str) -> None: """Keep every deletion check bound to the originally approved configured identity.""" _validate_depot_user(name) if configured_depot_user() != name: raise AccountRefusal("Konfigurierter Depotbenutzer geändert; keine Kontolöschung.") def account_inventory(name: str) -> dict[str, Any]: """Inventory local account dependencies without reading authentication material.""" if name != "adminuser": _require_configured_depot_user(name) try: account = pwd.getpwnam(name) except KeyError: return {"exists": False, "name": name} dependencies: list[str] = [] for entry in Path("/proc").iterdir(): if entry.name.isdigit(): try: if entry.stat().st_uid == account.pw_uid: dependencies.append(msg("laufender Prozess {pid}", pid=entry.name)) except FileNotFoundError: pass for root in (Path("/etc/systemd"), Path("/etc/cron.d"), Path("/etc/sudoers.d"), Path("/etc/pam.d")): for path in root.rglob("*"): if path.is_file() and not path.is_symlink(): try: if re.search(r"(? bool: """Require full WebDAV and an unused, unique local identity matching the configured target.""" if name is None: name = configured_depot_user() else: _require_configured_depot_user(name) lines = ACCOUNT_FILES["passwd"].read_text().splitlines() records = [line.split(":") for line in lines if line.startswith(name + ":")] try: account = pwd.getpwnam(name) except KeyError: if records: raise AccountRefusal("Depotbenutzer bleibt erhalten: lokaler Kontozustand nicht eindeutig.") from None _require_configured_depot_user(name) return False if not records: raise AccountRefusal("Depotbenutzer ist kein lokales Benutzerkonto (z. B. Domänenkonto); keine Löschung.") if not depot_webdav_ready(): raise AccountRefusal("Depotbenutzer bleibt erhalten: Depotprotokolle nicht vollständig WebDAV.") if len(records) != 1 or len(records[0]) != 7 or account.pw_uid <= 0 or account.pw_gid <= 0: raise AccountRefusal("Nur ein eindeutig lokales, unprivilegiertes Depotkonto darf entfernt werden.") local = records[0] if not local[2].isascii() or not local[2].isdigit() or not local[3].isascii() or not local[3].isdigit(): raise AccountRefusal("Ungültige lokale Depotkonto-Identität; keine Löschung.") if (int(local[2]), int(local[3]), local[5], local[6]) != ( account.pw_uid, account.pw_gid, account.pw_dir, account.pw_shell, ): raise AccountRefusal("Depotbenutzer bleibt erhalten: NSS-Identität weicht vom lokalen Konto ab.") if any(len(line.split(":")) != 7 or re.fullmatch(r"[0-9]+", line.split(":")[2]) is None for line in lines): raise AccountRefusal("Lokale Kontendaten nicht eindeutig; keine Löschung.") if sum(int(line.split(":")[2]) == account.pw_uid for line in lines) != 1: raise AccountRefusal("Geteilte UID; Depotbenutzer bleibt erhalten.") # NSS may also use winbind/LDAP; only the verified local identity is eligible. state = account_inventory(name) if state.get("owned_count", 0): raise AccountRefusal("Depotbenutzer besitzt Dateien; Dateibesitz verhindert die Kontolöschung.") if state.get("dependencies"): raise AccountRefusal("Depotbenutzer hat laufende oder konfigurierte Abhängigkeiten; keine Kontolöschung.") pattern = r"(? tuple[list[str], list[str], list[str]]: """Validate the complete saved identity before any restoration tool can run.""" try: fields = state["passwd"].split(":") shadow = state["shadow"].split(":") groups = state["groups"] if len(fields) != 7 or len(shadow) != 9 or not isinstance(groups, list): raise ValueError name = _validate_depot_user(fields[0]) if shadow[0] != name: raise ValueError if any(re.search(r"[\x00-\x1f\x7f]", value) for value in fields + shadow): raise ValueError if any(re.fullmatch(r"[0-9]+", fields[index]) is None or int(fields[index]) <= 0 for index in (2, 3)): raise ValueError if any(value and re.fullmatch(r"-?[0-9]+", value) is None for value in shadow[2:]): raise ValueError for record in groups: parts = record.split(":") if len(parts) != 4 or re.search(r"[\x00-\x1f\x7f]", record): raise ValueError if re.fullmatch(r"[A-Za-z_][A-Za-z0-9_-]*\$?", parts[0]) is None or len(parts[0]) > 32: raise ValueError if re.fullmatch(r"[0-9]+", parts[2]) is None: raise ValueError if not any(record.split(":")[2] == fields[3] for record in groups): raise ValueError except (KeyError, TypeError, AttributeError, ValueError, AccountRefusal): raise Refusal("Ungültige Kontosicherung.") from None return fields, shadow, groups def depot_user_restore(directory: Path) -> None: """Restore only the saved account and memberships, never overwrite whole account databases.""" path = directory / "account.json" if directory.is_symlink() or path.is_symlink(): raise Refusal("Unsichere Kontosicherung.") info = path.stat() if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o600: raise Refusal("Ungeschützte Kontosicherung.") try: state = json.loads(path.read_text()) except (ValueError, UnicodeError, OSError): raise Refusal("Kontosicherung nicht lesbar oder ungültig.") from None fields, shadow, groups = _validated_depot_backup(state) username = fields[0] for record in groups: name, _, gid, _ = record.split(":") try: group = grp.getgrnam(name) except KeyError: run(["groupadd", "--gid", gid, "--", name]) else: if group.gr_gid != int(gid): raise Refusal("Gruppen-ID inzwischen geändert; Wiederherstellung verweigert.") try: account = pwd.getpwnam(username) except KeyError: try: pwd.getpwuid(int(fields[2])) except KeyError: pass else: raise Refusal("Gesicherte UID inzwischen vergeben.") from None run( [ "useradd", "--no-create-home", "--no-user-group", "--uid", fields[2], "--gid", fields[3], "--home-dir", fields[5], "--shell", fields[6], "--comment", fields[4], username, ] ) else: if (account.pw_uid, account.pw_gid, account.pw_dir, account.pw_shell) != ( int(fields[2]), int(fields[3]), fields[5], fields[6], ): raise Refusal("Depotbenutzer inzwischen verändert; Wiederherstellung verweigert.") # Password material only through stdin, never command arguments or output. run(["chpasswd", "--encrypted"], data=f"{username}:{shadow[1]}\n") for option, value in zip(("-d", "-m", "-M", "-w", "-i", "-E"), shadow[2:8], strict=True): run(["chage", option, value or "-1", username]) for record in groups: name, _, _, members = record.split(":") if username in members.split(","): run(["gpasswd", "--add", username, name]) def depot_user_remove(name: str | None = None) -> None: """Remove the checked local identity without account backup or automatic recovery.""" if name is None: name = configured_depot_user() _require_configured_depot_user(name) def identity(username: str) -> tuple[int, int, str, str] | None: """Read only local/NSS identity attributes, never shadow or password fields.""" records = [ line.split(":") for line in ACCOUNT_FILES["passwd"].read_text().splitlines() if line.startswith(username + ":") ] try: account = pwd.getpwnam(username) except KeyError: if records: raise AccountRefusal("Lokaler Kontozustand nicht eindeutig; keine Kontolöschung.") from None return None if len(records) != 1 or len(records[0]) != 7: raise AccountRefusal("Depotbenutzer ist kein eindeutig lokales Benutzerkonto; keine Kontolöschung.") if account.pw_uid <= 0 or account.pw_gid <= 0: raise AccountRefusal("Nur ein unprivilegiertes Depotkonto darf entfernt werden.") fields = records[0] if any(re.fullmatch(r"[0-9]+", fields[index]) is None for index in (2, 3)): raise AccountRefusal("Ungültige lokale Depotkonto-Identität; keine Kontolöschung.") local = (int(fields[2]), int(fields[3]), fields[5], fields[6]) if local != (account.pw_uid, account.pw_gid, account.pw_dir, account.pw_shell): raise AccountRefusal("NSS-Identität weicht vom lokalen Konto ab; keine Kontolöschung.") return local approved_identity = identity(name) if not depot_user_removal_check(name): output("Depotbenutzer bereits entfernt; keine Änderung.") return if approved_identity is None or identity(name) != approved_identity: raise AccountRefusal("Depotkonto während Prüfung geändert; keine Kontolöschung.") # Only group metadata is kept in memory: userdel may remove a private primary group. groups: list[tuple[str, str]] = [] for line in ACCOUNT_FILES["group"].read_text().splitlines(): fields = line.split(":") if ( len(fields) != 4 or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_-]*\$?", fields[0]) is None or len(fields[0]) > 32 or re.fullmatch(r"[0-9]+", fields[2]) is None ): raise AccountRefusal("Lokale Gruppendaten nicht eindeutig; keine Kontolöschung.") if int(fields[2]) == approved_identity[1] or name in fields[3].split(","): groups.append((fields[0], fields[2])) if sum(int(gid) == approved_identity[1] for _, gid in groups) != 1: raise AccountRefusal("Primäre Gruppe nicht eindeutig lokal vorhanden; keine Kontolöschung.") _require_configured_depot_user(name) if not depot_user_removal_check(name): raise AccountRefusal("Kontozustand während Prüfung geändert; keine Kontolöschung.") if identity(name) != approved_identity: raise AccountRefusal("Depotkonto während Prüfung geändert; keine Kontolöschung.") _require_configured_depot_user(name) try: run(["userdel", name]) try: pwd.getpwnam(name) except KeyError: pass else: raise Refusal("Depotbenutzer-Löschung nicht wirksam.") # userdel may remove an empty private primary group; retain its identity. for group_name, gid in groups: try: group = grp.getgrnam(group_name) except KeyError: run(["groupadd", "--gid", gid, "--", group_name]) else: if group.gr_gid != int(gid): raise Refusal("Gruppenidentität inzwischen verändert.") except (Refusal, OSError, ValueError, KeyError, TypeError, subprocess.TimeoutExpired, KeyboardInterrupt): raise AccountRefusal( "Kontolöschung oder Gruppenprüfung fehlgeschlagen; kein automatischer Konto-Rollback. " "Keine separate Kontosicherung; manueller Abgleich von Konto und Gruppen erforderlich." ) from None output( "Depotbenutzer entfernt; Home, Depotdaten und Gruppen erhalten. Keine Kontosicherung oder automatische Wiederherstellung." ) def pcpatch_removal_check() -> bool: """Compatibility wrapper checking the configured depot user, without a pcpatch fallback.""" return depot_user_removal_check() def pcpatch_remove(directory: Path | None = None) -> None: """Remove the configured depot user; an old backup-directory argument is ignored.""" depot_user_remove() def pcpatch_restore(directory: Path) -> None: """Compatibility wrapper restoring the backup identity independently of live configuration.""" depot_user_restore(directory) def redis_connection_url(password: str) -> str: """Use URI userinfo so both Redis authentication and opsiconfd's secret filter see the password.""" return "unix://default:" + quote(password, safe="") + "@" + SOCKET + "?db=0" def redis_redaction_fix(directory: Path) -> None: """Repair this tool's legacy query URI in memory without rotating or displaying its password.""" overrides_clear(redis=True) config = effective_config() url = urlsplit(config.get("redis_internal_url", "")) if url.scheme != "unix" or url.path != SOCKET: raise Refusal("Nicht die von diesem Werkzeug eingerichtete Redis-Konfiguration.") password = parse_qs(url.query).get("password", [None])[0] or (unquote(url.password) if url.password else None) if not password: raise Refusal("Bestehendes Redis-Passwort nicht verfügbar; unverändert.") new_url = redis_connection_url(password) backup_file(OPSI_FILE, directory) backup_file(OPSI_DROPIN, directory) text = edit_config(OPSI_FILE.read_text(), {"redis-internal-url": new_url}) try: atomic_write(OPSI_FILE, text, mode=0o640, uid=0, gid=grp.getgrnam("opsiadmin").gr_gid) atomic_write(OPSI_DROPIN, OPSI_DROPIN_TEXT, mode=0o644) run(["systemctl", "daemon-reload"]) run(["systemctl", "restart", "opsiconfd"]) wait_ready() redis_validate() except (Refusal, OSError, ValueError, KeyboardInterrupt): restore_files(directory) run(["systemctl", "daemon-reload"]) run(["systemctl", "restart", "opsiconfd"]) wait_ready() raise Refusal("URI-Reparatur fehlgeschlagen; ursprüngliche Konfiguration wiederhergestellt.") from None output("URI-Logredaktion korrigiert, bestehendes Passwort erhalten; Socket/API geprüft.") # Never emit matching log lines; only report whether any raw password was found. leaked = False for path in Path("/var/log/opsi/opsiconfd").glob("*.log*"): if path.is_file(): with path.open("rb") as stream: for line in stream: if password.encode() in line: leaked = True journal = run(["journalctl", "-u", "opsiconfd", "-u", "redis-server", "--no-pager"]) leaked = leaked or password in journal if leaked: raise Refusal( "Unmaskiertes Redis-Passwort in Dienstlogs gefunden; sichere Bereinigung und ausdrückliche Rotation erforderlich." ) output( "Dienstlogs und Journal geprüft: kein unmaskiertes Redis-Passwort gefunden (nur Wahrheitswert, keine Logzeilen ausgegeben)." ) def display_inventory(operation: str, state: dict[str, Any], *, verbose: bool = False) -> None: """Show concise filtered findings, retaining precise override targets in both modes.""" if verbose: print(json.dumps(state, indent=2, ensure_ascii=False)) return if operation == "inventory": print(msg("Administrative API: ") + msg("verfügbar" if state.get("admin_api") is True else "nicht bestätigt")) features = state.get("config", {}).get("disabled_features", []) print(msg("Deaktivierte Funktionen: ") + (", ".join(features) if features else msg("keine"))) modules = state.get("modules", {}) print( msg("Lizenzen: ") + "; ".join(f"{name}: {msg('verfügbar' if value else 'fehlt')}" for name, value in modules.items()) ) elif operation == "depot-inventory": for obj in state.get("configs", []): print(f"{obj['id']}: {', '.join(obj.get('defaultValues', [])) or msg('nicht gesetzt')}") states = state.get("states", []) output(msg("Explizite Overrides: {count}", count=len(states))) for obj in states: print(f" {obj['configId']} / {obj['objectId']}: {', '.join(obj['values']) or msg('leer')}") depots = state.get("depots", []) output( msg( "Depots mit WebDAV-URL: {ready}/{count}", ready=sum(obj.get("webdav") is True for obj in depots), count=len(depots), ) ) output( msg( "Produkte: {products}; offene Aktionen: {actions}", products=len(state.get("products", [])), actions=len(state.get("active_actions", [])), ) ) elif operation == "redis-inventory": print( "Redis: " + msg("bereits abgesichert" if state.get("already_secure") is True else "noch nicht abgesichert") ) elif operation == "account": print( f"{state['name']}: {msg('vorhanden; Bestandsaufnahme ohne Löschung' if state.get('exists') else 'nicht vorhanden')}" ) else: raise Refusal("Unbekannte Bestandsanzeige.") def cli(argv: list[str] | None = None, *, verbose: bool = False) -> int: """Dispatch secret-safe operations from the wizard or recovery command line.""" with framework_translation([argparse]): parser = argparse.ArgumentParser( description=framework_msg("Secret-safe data operations for the opsi-secure wizard (standard library only).") ) parser.add_argument( "operation", choices=[ "inventory", "depot-inventory", "depot-apply", "redis-inventory", "redis-apply", "features", "feature-apply", "validate", "restore-files", "restore-depot", "account", "redis-redaction-fix", ], ) parser.add_argument("arguments", nargs="*") with framework_translation([argparse]): args = parser.parse_args(argv) try: if args.operation == "inventory": config = effective_config() license_info = api("backend_getLicensingInfo", False, False, False, False) display_inventory( "inventory", { "config": { key: config.get(key) for key in ( "run_as_user", "disabled_features", "multi_factor_auth", "disabled_auth_methods", ) }, "modules": { key: license_info.get("modules", {}).get(key, {}).get("available", False) for key in ("opsi_auth", "2fa", "vpn") }, "admin_api": api("accessControl_userIsAdmin"), }, verbose=verbose, ) elif args.operation == "depot-inventory": display_inventory(args.operation, depot_inventory(), verbose=verbose) elif args.operation == "depot-apply": depot_apply(Path(args.arguments[0]), delete_overrides="delete-overrides" in args.arguments[1:]) elif args.operation == "redis-inventory": display_inventory(args.operation, redis_inventory(), verbose=verbose) elif args.operation == "redis-apply": redis_apply(Path(args.arguments[0])) elif args.operation == "features": overrides_clear() print(json.dumps(effective_config().get("disabled_features", []))) elif args.operation == "feature-apply": feature_apply(args.arguments[1:], Path(args.arguments[0])) elif args.operation == "validate": if not service_ready(): raise Refusal("opsiconfd/JSON-RPC nicht verfügbar.") output("opsiconfd und frische administrative JSON-RPC-Anmeldung funktionieren.") if urlsplit(effective_config().get("redis_internal_url", "")).scheme == "unix": redis_validate() output("Redis-Socket, Authentifizierung, NOAUTH und fehlende TCP/TLS-Listener geprüft.") elif args.operation == "restore-files": restore_files(Path(args.arguments[0])) elif args.operation == "restore-depot": directory = Path(args.arguments[0]) api("config_updateObjects", protocol_objects(json.loads((directory / "configs.json").read_text()))) states = protocol_objects(json.loads((directory / "states.json").read_text()), states=True) if states: api("configState_createObjects", states) elif args.operation == "account": display_inventory(args.operation, account_inventory(args.arguments[0]), verbose=verbose) elif args.operation == "redis-redaction-fix": redis_redaction_fix(Path(args.arguments[0])) except (Refusal, OSError, ValueError, KeyError, IndexError, TypeError, KeyboardInterrupt): # Never stringify arbitrary errors: paths, URI query strings and response bodies can contain secrets. exc = sys.exception() if isinstance(exc, KeyboardInterrupt) and argv is not None: raise output( str(exc) if isinstance(exc, Refusal) else "Prüfung fehlgeschlagen; sensible Fehlerdetails unterdrückt.", file=sys.stderr, ) return 20 return 0 return _HelperNamespace(**locals()) helper = _bundled_helper() """Interactive, secret-safe hardening of package-based opsi 4.3 on Debian 13.""" import fcntl import inspect import json import os import re import shlex import shutil import signal import stat import subprocess import sys import tempfile from contextlib import contextmanager from pathlib import Path from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from collections.abc import Generator, Sequence from types import ModuleType import click OPSI_DIRECTORY = Path("/etc/opsi") BACKUP_ROOT = Path("/var/backups/opsi-secure") SAMBA_UNIT = re.compile(r"(?:smbd|nmbd|samba-ad-dc)\.(?:service|socket|timer)") ENABLED_STATES = {"enabled", "enabled-runtime", "disabled", "masked", "static", "indirect"} ACTIVE_STATES = {"active", "inactive", "failed"} VERSION = "0.1.0" YES_HELP = ( "ALLE Fragen mit Ja beantworten, einschließlich Backup, Risiken, destruktiver Kontolöschung, " "Dienständerungen und Wiederherstellung; ohne direktes Terminal. Bei check/Dry-Run inaktiv." ) def say(text: str) -> None: """Print only explicitly selected, non-secret information.""" text = helper.msg(text) if text.startswith("=== ") and text.endswith(" ==="): print(f"\n{'─' * 72}\n{text[4:-4]}\n{'─' * 72}") else: print(text) def terminal_available() -> bool: """Require direct input and output terminals, never authorize through a pipe.""" return sys.stdin.isatty() and sys.stdout.isatty() def tty_answer(prompt: str) -> str: """Read from the controlling terminal, defaulting to refusal on EOF/errors.""" if not terminal_available(): return "" try: # Separate streams avoid BufferedRandom's seek requirement on a terminal. with open("/dev/tty", "r", encoding="utf-8") as reader, open("/dev/tty", "w", encoding="utf-8") as writer: writer.write(prompt) writer.flush() return reader.readline().removesuffix("\n") except OSError: return "" def command_state(args: list[str]) -> tuple[int, str]: """Capture systemd states even when its exit code denotes inactive or disabled.""" env = helper.subprocess_environment() try: result = subprocess.run(args, capture_output=True, text=True, timeout=45, env=env, check=False) except (OSError, subprocess.TimeoutExpired): raise helper.Refusal("Dienstzustand nicht lesbar; Rohdaten unterdrückt.") from None return result.returncode, result.stdout.strip() def lock_directory(directory: Path = OPSI_DIRECTORY) -> int: """Exclusively lock an existing directory inode without creating a lock file.""" fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BaseException: os.close(fd) raise return fd def protected_backup(directory: Path, *, files_only: bool = False) -> bool: """Reject symlinked or unprotected backup paths before restoration.""" try: relative = directory.relative_to(BACKUP_ROOT) if not relative.parts or not relative.parts[0].startswith("run-"): return False if len(relative.parts) != (2 if files_only else 1): return False if files_only and relative.parts[-1] not in {"features", "redis"}: return False paths = [BACKUP_ROOT] + [BACKUP_ROOT.joinpath(*relative.parts[:i]) for i in range(1, len(relative.parts) + 1)] for path in paths: info = path.lstat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700: return False return True except (OSError, ValueError): return False def samba_records(path: Path) -> list[tuple[str, str, str]]: """Validate every saved Samba state before performing any service change.""" if path.is_symlink() or not path.is_file(): raise helper.Refusal("Ungültige Samba-Sicherung.") records: list[tuple[str, str, str]] = [] for line in path.read_text().splitlines(): fields = line.split("\t") if len(fields) != 3: raise helper.Refusal("Ungültige Samba-Sicherung.") unit, enabled, active = fields if not SAMBA_UNIT.fullmatch(unit) or enabled not in ENABLED_STATES or active not in ACTIVE_STATES: raise helper.Refusal("Nicht unterstützter Samba-Dienstzustand.") if any(record[0] == unit for record in records): raise helper.Refusal("Doppelte Samba-Einheit in Sicherung.") records.append((unit, enabled, active)) return records def restore_samba(path: Path) -> None: """Reinstate saved enablement and activity without activating originally inactive units.""" for unit, enabled, active in samba_records(path): if enabled in {"enabled", "enabled-runtime"}: helper.run(["systemctl", "enable", *(["--runtime"] if enabled == "enabled-runtime" else []), unit]) if active == "active": helper.run(["systemctl", "start", unit]) class BackupDeclined(Exception): """Stop all later measures when the mandatory baseline backup is declined.""" class BackupFailed(Exception): """Propagate baseline backup failures to the sanitized outer CLI boundary.""" class Wizard: """Coordinate independently approved measures and retain failures over later skips.""" def __init__(self, *, dry_run: bool = False, verbose: bool = False, assume_yes: bool = False) -> None: self.dry_run = dry_run self.verbose = verbose self.assume_yes = assume_yes and not dry_run self.samba_inventory_ready = True self.backup: Path | None = None self.backup_ready = False self.backup_planned = False self.result = 0 self.accepted_losses: list[str] = [] self.smb_risk_explained = False self.lock_fd: int | None = None def skipped(self) -> None: """Mark an incomplete run without hiding an earlier failed measure.""" if self.result != 20: self.result = 30 def confirm(self, prompt: str) -> bool: """Audit explicit automatic approval or read default-No consent, never in dry-run.""" if self.dry_run: return False prompt = helper.msg(prompt) if self.assume_yes: say(prompt + helper.msg(" [automatisch: Ja]")) return True return tty_answer(prompt + helper.msg(" [j/N]: ")).strip().casefold() in {"j", "ja", "y", "yes"} def warn_automatic_consent(self) -> None: """Warn before backup or restoration when all decisions are preapproved.""" if self.assume_yes and not self.dry_run: say( "WARNUNG: Automatische Zustimmung aktiv: ALLE Fragen werden mit Ja beantwortet, einschließlich " "initialer Sicherung, WebDAV-/SMB-Risiken, Override-Löschung, Dienständerungen, " "Funktionsabschaltungen, Kontolöschung des konfigurierten Depotbenutzers und Wiederherstellung. " "Kontolöschung ohne separate Kontosicherung oder automatischen Konto-Rollback; " "Wiederherstellung kann neuere Änderungen überschreiben. Schutzprüfungen bleiben verpflichtend." ) def risk_confirm(self, measure: str) -> bool: """Approve verified compatibility and known SMB risks with an ordinary yes/no.""" if self.dry_run: return False if not self.smb_risk_explained: say( "RISIKO: Capture/Clonezilla können Schreibzugriff verlieren; externe SMB-Nutzung und Automatisierung " "können ausfallen. Auch inaktive SMB-Abhängigkeiten sind nicht sicher ausschließbar." ) self.smb_risk_explained = True if measure == "WebDAV": action = helper.msg( "sollen jetzt beide Depot-Standardwerte ({ids}) auf WebDAV umgestellt werden", ids=", ".join(helper.PROTOCOL_IDS), ) elif measure == "WebDAV-Overrides": action = helper.msg( "sollen explizite Overrides beider Konfigurationen gesichert und gelöscht werden " "(Depot und/oder Client Werte; {ids})", ids=", ".join(helper.PROTOCOL_IDS), ) else: action = helper.msg("soll jetzt die Samba-Deaktivierung geprüft werden") if not self.confirm( helper.msg( "Sind Client-/Netboot-Komponenten und relevante Produkte für WebDAV geprüft, " "werden diese Risiken akzeptiert und {action}?", action=action, ) ): return False self.accepted_losses.append( helper.msg("{measure}: Kompatibilität geprüft und SMB-Risiko akzeptiert", measure=measure) ) return True def new_backup(self) -> Path: """Create one root-only backup run lazily, never during a dry-run.""" if self.dry_run: raise helper.Refusal("Backup im Dry-Run verweigert.") if self.backup is not None: return self.backup if BACKUP_ROOT.is_symlink(): raise helper.Refusal("Unsicheres Backup-Verzeichnis.") BACKUP_ROOT.mkdir(mode=0o700, parents=True, exist_ok=True) info = BACKUP_ROOT.stat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700: raise helper.Refusal("Unsicheres Backup-Verzeichnis.") self.backup = Path(tempfile.mkdtemp(prefix="run-", dir=BACKUP_ROOT)) say(helper.msg("Geschützte Sicherung: {backup}", backup=self.backup)) return self.backup def backup_step(self) -> bool: """Approve and create the baseline backup before inventory or measure decisions.""" if self.dry_run: say("PLAN: Vor den Maßnahmen ein opsiconfd-Backup in einem geschützten Verzeichnis erstellen.") self.backup_planned = True return True say("=== Initiale Sicherung ===") say("Das opsiconfd-Backup enthält die opsi-Konfiguration, das Backend und Redis-Daten.") say("Linux-Konten aus passwd/shadow werden durch das opsiconfd-Backup nicht gesichert.") say( "Eine kurze Wartungsunterbrechung ist möglich; Ablage in einem geschützten, nur für root zugänglichen Verzeichnis." ) if not self.confirm( "Vor den Maßnahmen ein opsiconfd-Backup erstellen? Der Server wird dafür in den Maintenance Mode (Wartungsmodus) versetzt." ): say("Keine Zustimmung zur initialen Sicherung; keine Maßnahmen ausgeführt.") self.skipped() return False helper.opsi_backup(self.new_backup()) return True def prepare_change(self) -> None: """Require one successful baseline backup per run, aborting on refusal or failure.""" if self.dry_run: raise helper.Refusal("Änderungsvorbereitung im Dry-Run verweigert.") if self.backup_ready: return try: approved = self.backup_step() except Exception: # noqa: BLE001 -- no backup failure may fall through to later measures raise BackupFailed(helper.msg("Initiale Sicherung fehlgeschlagen.")) from None if not approved: self.skipped() raise BackupDeclined self.backup_ready = True def plan_change(self, text: str) -> None: """Describe a pending dry-run measure and its once-per-run baseline prerequisite.""" if not self.backup_planned: say("PLAN: Vor den Maßnahmen ein opsiconfd-Backup in einem geschützten Verzeichnis erstellen.") self.backup_planned = True say(text) def prerequisites(self, *, files_only: bool = False) -> int: """Check root/target prerequisites and retain the directory lock for the whole run.""" if os.geteuid() != 0: say("Root-Rechte erforderlich; keine Änderung.") return 2 if not files_only: identity = {} try: for line in Path("/etc/os-release").read_text().splitlines(): if "=" in line and not line.startswith("#"): key, value = line.split("=", 1) words = shlex.split(value) identity[key] = words[0] if words else "" except (OSError, ValueError): return 2 if identity.get("ID") != "debian" or identity.get("VERSION_ID") != "13": say("Nur Debian 13 freigegeben.") return 2 for tool in ("opsiconfd", "opsi-cli", "systemctl", "ss", "dpkg-query", "find"): if not shutil.which(tool): say(helper.msg("Erforderliches Werkzeug fehlt: {tool}", tool=tool)) return 2 try: self.lock_fd = lock_directory(OPSI_DIRECTORY) except BlockingIOError: say("Andere opsi-secure-Ausführung aktiv.") return 10 except OSError: return 2 if files_only: return 0 try: version = helper.run(["dpkg-query", "-W", "-f=${Version}", "opsiconfd"]).strip() if not version.startswith("4.3."): return 2 helper.run(["dpkg-query", "-W", "opsiconfd", "opsi-utils", "redis-server", "opsi-linux-bootimage"]) helper.run(["opsi-cli", "--version"]) except helper.Refusal: return 2 say("Paketbasierte opsi-4.3-/Redis-/Bootimage-Installation auf Debian 13 geprüft.") return 0 def inventory_cli(self, operation: list[str]) -> int: """Request filtered details only when enabled, preserving default helper calls.""" if self.verbose: return helper.cli(operation, verbose=True) return helper.cli(operation) def inventory(self) -> None: """Display helper-filtered inventory and only share names/paths from Samba.""" say("=== Bestandsaufnahme (keine Konfigurationsänderung) ===") if self.inventory_cli(["inventory"]): raise helper.Refusal("Bestandsaufnahme fehlgeschlagen.") for title, operation in ( ("Depotprotokolle, Overrides und Produkte", ["depot-inventory"]), ("Redis", ["redis-inventory"]), ("Konto adminuser", ["account", "adminuser"]), ): if self.verbose: say(f"=== {helper.msg(title)} ===") if self.inventory_cli(operation): self.skipped() try: name = helper.configured_depot_user() if self.verbose: say(helper.msg("=== Konto {name} ===", name=name)) if self.inventory_cli(["account", name]): self.skipped() except helper.AccountRefusal as error: say(str(error)) self.skipped() except (helper.Refusal, OSError, ValueError, KeyError, TypeError): say("Depotbenutzer-Bestandsaufnahme fehlgeschlagen; sensible Fehlerdetails unterdrückt.") self.skipped() if self.verbose: say("=== Samba-Einheiten und Freigaben ===") try: units = self.samba_units() if self.verbose: for unit, state in units: say(f"{unit} {state}") else: say( helper.msg("Samba-Einheiten: ") + (", ".join(f"{unit} ({state})" for unit, state in units) or helper.msg("keine")) ) except (helper.Refusal, OSError, ValueError): say("WARNUNG: Samba-Einheiten konnten nicht gelesen werden; Samba-Deaktivierung gesperrt.") self.samba_inventory_ready = False self.skipped() try: if not shutil.which("testparm"): raise helper.Refusal("testparm nicht verfügbar.") lines = helper.run(["testparm", "-s"]).splitlines() for line in lines if self.verbose else []: if re.match(r"^\[[^\r\n]*\]$|^\s*path =", line): say(line) if not self.verbose: shares = sum(bool(re.fullmatch(r"\[[^\r\n]*\]", line)) and line != "[global]" for line in lines) say( helper.msg( "Samba-Konfiguration lesbar; Freigaben: {shares} (Details mit --verbose).", shares=shares ) ) except (helper.Refusal, OSError, ValueError): say("WARNUNG: Samba-Freigaben konnten mit testparm nicht gelesen werden; Samba-Deaktivierung gesperrt.") self.samba_inventory_ready = False self.skipped() def depot_step(self) -> None: """Approve only pending defaults or explicit override deletion using one inventory read.""" say("=== WebDAV / Samba ===") state = helper.depot_inventory() helper.display_inventory("depot-inventory", state, verbose=self.verbose) defaults_pending = any(obj.get("defaultValues") != ["webdav"] for obj in state["configs"]) overrides_pending = bool(state["states"]) if not defaults_pending and not overrides_pending: say("WebDAV-Standardwerte bereits gesetzt; keine expliziten Overrides, keine Änderung.") return if self.dry_run: if defaults_pending: self.plan_change( "PLAN: Beide WebDAV-Standardwerte nach gemeinsamer Kompatibilitäts-/Risikobestätigung umstellen." ) if overrides_pending: self.plan_change("PLAN: Explizite Overrides nur nach separater Zustimmung sichern und löschen.") return if defaults_pending: say("Beide Standardwerte auf webdav; alle übrigen Definitionseigenschaften bleiben erhalten.") if not self.risk_confirm("WebDAV" if defaults_pending else "WebDAV-Overrides"): say("WebDAV unverändert: keine Zustimmung.") self.skipped() return delete = overrides_pending and ( not defaults_pending or self.confirm( helper.msg( "Explizite Overrides beider Konfigurationen sichern und löschen (Depot und/oder Client Werte). " "Betroffene Konfigurationen: {ids}?", ids=", ".join(helper.PROTOCOL_IDS), ) ) ) if overrides_pending and not delete: say("Explizite Overrides bleiben erhalten; vollständige WebDAV-Umstellung nicht bestätigt.") self.skipped() self.prepare_change() helper.depot_apply(self.new_backup() / "depot", delete_overrides=delete) def redis_step(self) -> None: """Approve local-only use and socket/password migration in a single decision.""" say("=== Redis ===") try: state = helper.redis_inventory() except helper.Refusal: self.skipped() return helper.display_inventory("redis-inventory", state, verbose=self.verbose) if state.get("already_secure") is True: say("Bereits abgesichert und validiert; keine Sicherung, kein Passwortwechsel, kein Neustart.") return say("Lokaler Socket mit Passwort, TCP/TLS aus. Kein Passwort wird ausgegeben.") say("Kurze Unterbrechung: opsiconfd stoppen, Redis neu starten, opsiconfd starten.") if self.verbose: say("Includes, bestehende ACL/Passwörter, weitere Verbraucher und Overrides sperren die Migration.") say("Auch inaktive/periodische Verbraucher müssen geprüft sein; Socket 0660, RuntimeDirectory 0750.") if self.dry_run: self.plan_change( "PLAN: 256-Bit-Passwort, Socket 0660, RuntimeDirectory 0750, Dienstgruppe redis, TCP/TLS aus." ) return if not self.confirm( "Wird Redis ausschließlich für dieses lokale opsi verwendet und soll jetzt abgesichert werden?" ): self.skipped() return self.prepare_change() helper.redis_apply(self.new_backup() / "redis") def auth_step(self) -> None: """Explicitly refuse unverified administrator and 2FA migration.""" say("=== Administrator, 2FA und Aussperrungsschutz ===") say( "Ohne opsi_auth-Lizenz ist die automatische Migration auf einen alternativen Administrator nicht verfügbar." ) say("Ohne 2fa-Lizenz ist kein 2FA für den Administrator möglich.") say("Administratoranlage/2FA unsupported: Auch mit Lizenzen ist der Loginmechanismus hier nicht validiert.") say("Bestehende PAM-Administration behalten; unterstützten Datenbank-Workflow beim Hersteller klären.") say("Kein Secret erzeugt, kein TOTP zurückgesetzt, kein lokales adminuser-Konto gelöscht.") if self.verbose: say("Linux-Konto, User-Objekt, Passwort-Hash, Gruppenberechtigung und TOTP sind getrennte Mechanismen.") say("user_setCredentials betrifft NUR den Depotbenutzer, nicht einen neuen Administrator!") say("opsiconfd 4.3.59.1 benötigt opsi_auth für Datenbank-Anmeldung und 2fa für TOTP-Registrierung.") say( "totp_mandatory betrifft Passwort-Benutzer einschließlich Nichtadmins; Hostauth/Tokens haben Ausnahmen." ) self.skipped() def features_step(self) -> None: """Collect individual approvals and commit the merged selection exactly once.""" say("=== Nicht benötigte Funktionen ===") try: helper.overrides_clear() current = helper.effective_config().get("disabled_features", []) except helper.Refusal: self.skipped() return if not isinstance(current, list) or any(not isinstance(value, str) for value in current): raise helper.Refusal("Ungültige Funktionskonfiguration.") say( helper.msg( "Vorhandene disabled-features: {features}", features=", ".join(current) if current else helper.msg("keine"), ) ) if self.verbose: say(json.dumps(current, indent=2, ensure_ascii=False)) chosen = [] explanations = ( "Nicht authentifizierte /status-Seite aus; Status-Monitoring kann betroffen sein.", "Nur RPC-Werkzeug auf der Admin-Seite aus. Benötigte JSON-RPC-API /rpc bleibt an.", "Server-Prozessausführung über Messagebus aus; Remote-Prozessworkflows können ausfallen.", ) for feature, explanation in zip(helper.FEATURES, explanations, strict=True): if feature in current: say(helper.msg("{feature}: Bereits deaktiviert; keine Änderung.", feature=feature)) continue say(f"{feature}: {helper.msg(explanation)}") if self.dry_run: self.plan_change(helper.msg("PLAN: {feature} nach separater Zustimmung ergänzen.", feature=feature)) elif self.confirm( helper.msg( "Funktion {feature} deaktivieren? opsiconfd-Neustart am Ende dieser Auswahl erforderlich.", feature=feature, ) ): chosen.append(feature) if chosen: self.prepare_change() helper.feature_apply(chosen, self.new_backup() / "features") def samba_units(self) -> list[tuple[str, str]]: """Read only allowlisted Samba units and their installation states.""" units = [] for line in helper.run(["systemctl", "list-unit-files", "--no-pager", "--no-legend"]).splitlines(): fields = line.split() if len(fields) >= 2 and SAMBA_UNIT.fullmatch(fields[0]): units.append((fields[0], fields[1])) return units def samba_step(self) -> None: """Require verified WebDAV, prevalidate unit states and roll back failed disabling.""" if not self.samba_inventory_ready: say("ÜBERSPRUNGEN: Samba-Bestandsaufnahme unvollständig; Samba-Deaktivierung gesperrt.") self.skipped() return records = [] units = [] try: for unit, state in self.samba_units(): if state == "alias": continue _, enabled = command_state(["systemctl", "is-enabled", unit]) _, active = command_state(["systemctl", "is-active", unit]) if enabled not in ENABLED_STATES or active not in ACTIVE_STATES: say("Nicht unterstützter Dienstzustand; unverändert.") self.skipped() return if active == "active" or enabled in {"enabled", "enabled-runtime"}: units.append(unit) records.append(f"{unit}\t{enabled}\t{active}\n") except (helper.Refusal, OSError, ValueError): say("WARNUNG: Samba-Einheiten konnten nicht gelesen werden; Samba-Deaktivierung gesperrt.") self.samba_inventory_ready = False self.skipped() return if not units: say("Samba bereits deaktiviert oder keine Samba-Units vorhanden; keine Änderung.") return if self.dry_run: self.plan_change( "PLAN: Samba nur deaktivieren, wenn BEIDE Depotprotokolle samt Overrides effektiv WebDAV sind." ) say( " Separate Zustimmung zur Deaktivierung; Zustand sichern, nur ausstehende Units stoppen/deaktivieren." ) return if not helper.depot_webdav_ready(): say("ÜBERSPRUNGEN: Samba bleibt aktiv. Beide Depotprotokolle müssen vollständig auf WebDAV stehen.") say("Defaults, explizite Overrides und effektive Client-/Netboot-Werte werden gemeinsam geprüft.") self.skipped() return say("OK: Beide Depotprotokolle, Overrides und effektive Werte sind WebDAV.") webdav_approved = any(loss.startswith(("WebDAV:", "WebDAV-Overrides:")) for loss in self.accepted_losses) if not webdav_approved and not self.risk_confirm("Samba"): self.skipped() return if not self.confirm( "Samba-Dienste und deren Aktivierungseinheiten stoppen/deaktivieren (keine Pakete oder Nutzdaten löschen)?" ): self.skipped() return # Stop activators before services so they cannot reactivate a stopped daemon. units.sort(key=lambda unit: (unit.endswith(".service"), unit)) # Recheck after interactive decisions, immediately before any backup or service mutation. if not self.samba_inventory_ready: say("ÜBERSPRUNGEN: Samba-Bestandsaufnahme unvollständig; Samba-Deaktivierung gesperrt.") self.skipped() return if not helper.depot_webdav_ready(): say("ÜBERSPRUNGEN: Depotprotokolle inzwischen geändert; Samba bleibt unverändert.") self.skipped() return self.prepare_change() directory = self.new_backup() / "samba" directory.mkdir(mode=0o700) path = directory / "state.tsv" helper.atomic_write(path, "".join(records), mode=0o600) try: for unit in units: helper.run(["systemctl", "disable", "--now", unit]) for unit in units: code, active = command_state(["systemctl", "is-active", unit]) _, enabled = command_state(["systemctl", "is-enabled", unit]) if ( code == 0 or active not in {"inactive", "failed"} or enabled not in ENABLED_STATES - {"enabled", "enabled-runtime"} ): raise helper.Refusal("Samba-Deaktivierung nicht wirksam.") except (helper.Refusal, OSError, ValueError, KeyboardInterrupt): restore_samba(path) raise helper.Refusal("Samba-Änderung fehlgeschlagen; gespeicherte Zustände wiederhergestellt.") from None say("Samba-Einheiten gestoppt/deaktiviert; Freigaben und Nutzdaten erhalten. Kein SMB-Funktionstest.") def account_steps(self) -> None: """Pin the configured depot identity before checks and separate deletion approval.""" say("=== Benutzer ===") say("Lokales adminuser-Konto wird nicht gelöscht; bestehende PAM-Administration bleibt erhalten.") say( "Ohne opsi_auth-Lizenz ist die automatische Migration auf einen alternativen Administrator nicht verfügbar." ) say("Ohne 2fa-Lizenz ist kein 2FA für den Administrator möglich.") say( "Auch mit Lizenzen bleibt die Administrator-/2FA-Migration hier unsupported: Loginmechanismus nicht validiert." ) self.skipped() try: name = helper.configured_depot_user() say(helper.msg("Konfigurierter Depotbenutzer: {name}", name=name)) eligible = helper.depot_user_removal_check(name) except helper.AccountRefusal as error: say(str(error)) return except (helper.Refusal, OSError, ValueError, KeyError, TypeError): say("Depotbenutzer bleibt erhalten: Konfiguration, WebDAV oder Abhängigkeitsprüfung nicht freigegeben.") return if not eligible: say(helper.msg("{name} bereits entfernt; keine Änderung.", name=name)) return say( helper.msg( "{name}: Nach vollständiger WebDAV-Umstellung kann das lokale Konto entfernt werden. " "Home/Depotdaten und Gruppen bleiben erhalten.", name=name, ) ) say( "WARNUNG: Keine separate Kontosicherung und keine automatische Wiederherstellung des gelöschten Linux-Benutzers. " "Das opsiconfd-Backup sichert keine Linux-Konten aus passwd/shadow." ) if self.dry_run: self.plan_change( helper.msg( "PLAN: {name} nach WebDAV-/Abhängigkeitsprüfung und separater Zustimmung entfernen (ohne Kontosicherung).", name=name, ) ) return if self.confirm(helper.msg("Lokales Konto {name} jetzt entfernen?", name=name)): self.prepare_change() helper.depot_user_remove(name) def restore_step(self, directory: Path, *, files_only: bool = False) -> int: """Restore the saved account identity before protocols/services; reject ambiguous backups.""" if not protected_backup(directory, files_only=files_only): return 2 account = directory / "depot-user" legacy_account = directory / "pcpatch" if not files_only: for backup in (account, legacy_account): if backup.is_symlink(): raise helper.Refusal("Symlink in Wiederherstellung verweigert.") if account.exists() and legacy_account.exists(): raise helper.Refusal("Mehrdeutige Depotkonto-Sicherungen; keine Wiederherstellung.") if not account.exists(): account = legacy_account if account.exists(): info = account.lstat() if not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or stat.S_IMODE(info.st_mode) != 0o700: raise helper.Refusal("Ungeschützte Depotkonto-Sicherung.") self.warn_automatic_consent() if not self.confirm( "Geschützten Backup-Zustand wiederherstellen? Neuere Änderungen können verloren gehen; betroffene Dienste separat prüfen." ): return 0 if files_only: helper.restore_files(directory) say("Dateien wiederhergestellt; Dienste manuell prüfen/starten. Keine API-Prüfung oder Dienstneustarts.") return 0 if account.exists(): helper.depot_user_restore(account) for measure in ("features", "redis", "depot"): backup = directory / measure if backup.is_symlink(): raise helper.Refusal("Symlink in Wiederherstellung verweigert.") if not backup.is_dir(): continue if measure == "depot": if helper.cli(["restore-depot", str(backup)]): raise helper.Refusal("Depot-Wiederherstellung fehlgeschlagen.") continue helper.restore_files(backup) if measure == "redis": helper.run(["systemctl", "daemon-reload"]) helper.run(["systemctl", "restart", *(["redis-server"] if measure == "redis" else []), "opsiconfd"]) if (directory / "samba").is_symlink(): raise helper.Refusal("Symlink in Wiederherstellung verweigert.") state = directory / "samba/state.tsv" if state.exists() or state.is_symlink(): restore_samba(state) if helper.cli(["validate"]): raise helper.Refusal("Wiederherstellungsprüfung fehlgeschlagen.") say("Wiederherstellung geprüft soweit unterstützt; keine Transaktionsgarantie für API/Dateien/Dienste/Konten.") return 0 def execute(self) -> int: """Require the baseline first, then inventory and individually approved pending measures.""" self.warn_automatic_consent() if self.dry_run: self.backup_step() else: try: self.prepare_change() except BackupDeclined: return self.result self.inventory() say("=== Plan und Anwendung ===") for step in ( self.depot_step, self.samba_step, self.redis_step, self.auth_step, self.features_step, self.account_steps, ): try: step() except BackupDeclined: return self.result except BackupFailed: raise except helper.AccountRefusal as error: say(str(error)) self.result = 20 except (helper.Refusal, OSError, ValueError, KeyError, TypeError): say("Maßnahme fehlgeschlagen; Rohfehler/Geheimnisse werden nicht protokolliert.") self.result = 20 say("=== Abschlussprüfung ===") if helper.cli(["validate"]): self.result = 20 say("Konfigurations-/lokale Integrationstests sind keine realen Client-/Netboot-/2FA-Funktionstests.") if self.accepted_losses: say("Bewusst akzeptierte Funktionsverluste:") for loss in self.accepted_losses: say(loss) if self.backup: say( helper.msg( "Backups und Wiederherstellung: {backup} (root-only), --restore mit separater Bestätigung.", backup=self.backup, ) ) say( helper.msg( "Ergebniscode: {result} (30 = teilweise übersprungen, nicht vollständig gehärtet).", result=self.result ) ) return self.result def check(self) -> int: """Read inventory and validate without entering any hardening or approval step.""" self.inventory() say("=== Prüfung (keine Konfigurationsänderung) ===") if helper.cli(["validate"]): self.result = 20 return self.result def _interrupt(signum: int, frame: object) -> None: raise KeyboardInterrupt def run_wizard( command: str, *, dry_run: bool = False, directory: Path | None = None, verbose: bool = False, assume_yes: bool = False, ) -> int: """Run a selected operation with protected lifecycle and sanitized runtime errors.""" if dry_run and command in {"restore", "restore-files"}: say("Dry-Run mit Wiederherstellung verweigert; keine Änderung.") return 2 wizard = Wizard(dry_run=dry_run or command == "check", verbose=verbose, assume_yes=assume_yes) previous_handler = None previous_umask = None result = 20 try: previous_handler = signal.signal(signal.SIGTERM, _interrupt) previous_umask = os.umask(0o077) result = wizard.prerequisites(files_only=command == "restore-files") if result: pass elif not wizard.dry_run and not wizard.assume_yes and not terminal_available(): say("Anwendung nur in einem direkten interaktiven Terminal oder mit explizitem --yes/-y.") result = 2 elif directory is not None: result = wizard.restore_step(directory, files_only=command == "restore-files") elif command == "check": result = wizard.check() else: result = wizard.execute() except (Exception, KeyboardInterrupt): # noqa: BLE001 -- outer CLI boundary must suppress all raw secret-bearing errors helper.output( "Abbruch: Maßnahme fehlgeschlagen; Rohfehler/Geheimnisse werden nicht protokolliert.", file=sys.stderr ) result = 20 finally: # Attempt every cleanup even when another cleanup fails; failures outrank skips. for cleanup in ( lambda: os.close(wizard.lock_fd) if wizard.lock_fd is not None else None, lambda: signal.signal(signal.SIGTERM, previous_handler) if previous_handler is not None else None, lambda: os.umask(previous_umask) if previous_umask is not None else None, ): try: cleanup() except (Exception, KeyboardInterrupt): # noqa: BLE001 -- cleanup must not expose raw errors either helper.output("Abbruch: Aufräumen fehlgeschlagen; Rohfehler unterdrückt.", file=sys.stderr) result = 20 return result def _load_click() -> ModuleType: """Load optional Rich help lazily, falling back only on import failures.""" try: import rich_click as click except ImportError: import click return click click.rich_click.USE_RICH_MARKUP = True click.rich_click.MAX_WIDTH = 140 click.rich_click.STYLE_USAGE = "bold cyan3" click.rich_click.STYLE_OPTION = "bold cyan" click.rich_click.STYLE_SWITCH = "bold light_sea_green" click.rich_click.STYLE_METAVAR = "cyan3" return click @contextmanager def _click_translation() -> Generator[None]: """Localize framework labels only within this CLI and restore renderer settings.""" modules = [ sys.modules[name] for name in ( "click.core", "click.decorators", "click.exceptions", "click.types", "click.formatting", "click.parser", "rich_click.decorators", "rich_click.rich_help_rendering", ) if name in sys.modules ] config = sys.modules.get("rich_click.rich_click") labels = ( "ARGUMENTS_PANEL_TITLE", "OPTIONS_PANEL_TITLE", "COMMANDS_PANEL_TITLE", "ERRORS_PANEL_TITLE", "ABORTED_TEXT", ) saved = {name: getattr(config, name) for name in labels} if config else {} try: if config: for name, value in saved.items(): setattr(config, name, helper.framework_msg(value)) with helper.framework_translation(modules): yield finally: if config: for name, value in saved.items(): setattr(config, name, value) def create_cli() -> click.Group: """Create the command tree with optional Rich help and no target checks.""" click = _load_click() command_base = getattr(click, "RichCommand", click.Command) group_base = getattr(click, "RichGroup", click.Group) class LocalizedCommand(command_base): """Render and parse with scoped translation rather than global locale changes.""" def main(self, *args: Any, **kwargs: Any) -> Any: with _click_translation(): return super().main(*args, **kwargs) def get_help(self, ctx: click.Context) -> str: with _click_translation(): return super().get_help(ctx) def format_usage(self, ctx: click.Context, formatter: click.HelpFormatter) -> None: formatter.write_usage( ctx.command_path, " ".join(self.collect_usage_pieces(ctx)), prefix=helper.framework_msg("Usage:") + " " ) class LocalizedGroup(LocalizedCommand, group_base): """Apply the same localization to root help and all child commands.""" command_class = LocalizedCommand @click.group( cls=LocalizedGroup, name="opsi-secure", invoke_without_command=True, context_settings={"help_option_names": ["--help", "-h"]}, epilog="Exit-Codes: 0 geprüft/ausgeführt, 2 Argumente/Voraussetzungen, 10 parallele Ausführung, " "20 fehlgeschlagen, 30 teilweise übersprungen.", ) @click.option("--dry-run", is_flag=True, help="Keine Backups, Geheimnisse, Änderungen oder Neustarts.") @click.option( "--verbose", "-v", is_flag=True, help="Gefilterte JSON-Details und zusätzliche Erläuterungen anzeigen." ) @click.option("--yes", "-y", "assume_yes", is_flag=True, help=YES_HELP) @click.version_option( VERSION, prog_name="opsi-secure", message=helper.framework_msg("%(prog)s, version %(version)s") ) @click.pass_context def cli(ctx: click.Context, dry_run: bool, verbose: bool, assume_yes: bool) -> int | None: """Interaktive opsi-4.3-Härtung auf Debian 13. Root und paketbasierte Installation erforderlich. Ohne --yes/-y Bestätigungen nur am direkten Terminal, Voreinstellung Nein. Lesende API-Aufrufe können serverseitig Sitzungsdaten und Zugriffslogs erzeugen. Authentifizierungs-/2FA-Migration und adminuser-Löschung bleiben unsupported. Konfigurierten Depotbenutzer nur nach Schutzprüfung entfernen. """ ctx.obj = {"dry_run": dry_run, "verbose": verbose, "assume_yes": assume_yes} if ctx.invoked_subcommand is None: click.echo(ctx.get_help()) return 0 return None @cli.command() @click.option("--verbose", "-v", is_flag=True, help="Gefilterte JSON-Details anzeigen.") @click.pass_context def check(ctx: click.Context, verbose: bool) -> int: """Bestandsaufnahme und Validierung ohne Änderungen oder Bestätigungen.""" if verbose or ctx.obj["verbose"]: return run_wizard("check", dry_run=ctx.obj["dry_run"], verbose=True) return run_wizard("check", dry_run=ctx.obj["dry_run"]) @cli.command() @click.option("--dry-run", is_flag=True, help="Nur prüfen und Maßnahmen planen, ohne Bestätigungen.") @click.option( "--verbose", "-v", is_flag=True, help="Gefilterte JSON-Details und zusätzliche Erläuterungen anzeigen." ) @click.option("--yes", "-y", "assume_yes", is_flag=True, help=YES_HELP) @click.pass_context def harden(ctx: click.Context, dry_run: bool, verbose: bool, assume_yes: bool) -> int: """Maßnahmen einzeln bestätigen, anwenden und anschließend validieren.""" if assume_yes or ctx.obj["assume_yes"]: return run_wizard( "harden", dry_run=dry_run or ctx.obj["dry_run"], verbose=verbose or ctx.obj["verbose"], assume_yes=True ) if verbose or ctx.obj["verbose"]: return run_wizard("harden", dry_run=dry_run or ctx.obj["dry_run"], verbose=True) return run_wizard("harden", dry_run=dry_run or ctx.obj["dry_run"]) @cli.command() @click.option("--yes", "-y", "assume_yes", is_flag=True, help=YES_HELP) @click.option("--verbose", "-v", is_flag=True, help="Zusätzliche Erläuterungen anzeigen.") @click.argument("backup_directory", type=click.Path(path_type=Path)) @click.pass_context def restore(ctx: click.Context, backup_directory: Path, assume_yes: bool, verbose: bool) -> int: """Geschützten Lauf wiederherstellen, einschließlich API und Dienstneustarts.""" if assume_yes or ctx.obj["assume_yes"]: return run_wizard( "restore", dry_run=ctx.obj["dry_run"], directory=backup_directory, verbose=verbose or ctx.obj["verbose"], assume_yes=True, ) if verbose or ctx.obj["verbose"]: return run_wizard("restore", dry_run=ctx.obj["dry_run"], directory=backup_directory, verbose=True) return run_wizard("restore", dry_run=ctx.obj["dry_run"], directory=backup_directory) @cli.command("restore-files") @click.option("--yes", "-y", "assume_yes", is_flag=True, help=YES_HELP) @click.option("--verbose", "-v", is_flag=True, help="Zusätzliche Erläuterungen anzeigen.") @click.argument("backup_directory", type=click.Path(path_type=Path)) @click.pass_context def restore_files(ctx: click.Context, backup_directory: Path, assume_yes: bool, verbose: bool) -> int: """Geschütztes Maßnahmenbackup nur als Dateien wiederherstellen. API-unabhängig; Root und Verzeichnislock erforderlich; ohne --yes/-y direktes Terminal. Kein opsi-cli nötig. Keine Dienstneustarts; Dienste danach manuell prüfen/starten. """ if assume_yes or ctx.obj["assume_yes"]: return run_wizard( "restore-files", dry_run=ctx.obj["dry_run"], directory=backup_directory, verbose=verbose or ctx.obj["verbose"], assume_yes=True, ) if verbose or ctx.obj["verbose"]: return run_wizard("restore-files", dry_run=ctx.obj["dry_run"], directory=backup_directory, verbose=True) return run_wizard("restore-files", dry_run=ctx.obj["dry_run"], directory=backup_directory) for command in (cli, *cli.commands.values()): command.options_metavar = helper.framework_msg("[OPTIONS]") if command.help: command.help = helper.msg(inspect.cleandoc(command.help)) if command.epilog: command.epilog = helper.msg(command.epilog) for parameter in command.params: if isinstance(parameter, click.Option) and parameter.help: parameter.help = helper.framework_msg(helper.msg(parameter.help)) cli.subcommand_metavar = helper.framework_msg("[COMMAND] [ARGS]...") return cli def main(argv: Sequence[str] | None = None) -> int: """Parse commands while keeping dependency errors, aborts and runtime errors controlled.""" try: click = _load_click() cli = create_cli() except ImportError: helper.output( "Python-Laufzeitabhängigkeit fehlt: click erforderlich; für diesen Interpreter installieren.", file=sys.stderr, ) return 2 except Exception: # noqa: BLE001 -- import/configuration errors must not expose secrets or trigger fallback helper.output("Abbruch: CLI fehlgeschlagen; Rohfehler unterdrückt.", file=sys.stderr) return 20 args = list(sys.argv[1:] if argv is None else argv) if "--dry-run" in args and all(arg in {"--dry-run", "--verbose", "-v", "--yes", "-y"} for arg in args): args = ["harden", *args] else: args = [{"--restore": "restore", "--restore-files": "restore-files"}.get(arg, arg) for arg in args] try: return cli.main(args=args, prog_name="opsi-secure", standalone_mode=False) or 0 except ImportError: # Rich imports some rendering dependencies lazily, only when help is requested. helper.output("Python-CLI-Abhängigkeit fehlt; optionale rich-click-/rich-Installation prüfen.", file=sys.stderr) return 2 except click.ClickException as error: with _click_translation(): error.show() return error.exit_code except (click.Abort, KeyboardInterrupt): helper.output("Abbruch; Rohfehler unterdrückt.", file=sys.stderr) return 20 except Exception: # noqa: BLE001 -- never let unexpected parser/rendering errors disclose secrets helper.output("Abbruch: CLI fehlgeschlagen; Rohfehler unterdrückt.", file=sys.stderr) return 20 if __name__ == "__main__": raise SystemExit(main())