# user_control_agent.py import os import sys import subprocess import time import argparse import logging import psutil # Für is_user_logged_in und Prozessmanagement # Farben für CLI-Ausgabe (optional, aber nützlich für Debugging) class LogColors: INFO = "\033[94m" DEBUG = "\033[92m" WARNING = "\033[93m" ERROR = "\033[91m" RESET = "\033[0m" # Logging einrichten LOG_FILE = os.path.expanduser("~/safe_kiddo_user_control.log") logging.basicConfig( filename=LOG_FILE, level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) def log_message(level, message, debug_mode=False): """Loggt Nachrichten und gibt sie optional farbig auf der CLI aus.""" if level == "info": logging.info(message) if debug_mode: print(f"{LogColors.INFO}[INFO]{LogColors.RESET} {message}") elif level == "warning": logging.warning(message) if debug_mode: print(f"{LogColors.WARNING}[WARNING]{LogColors.RESET} {message}") elif level == "error": logging.error(message) if debug_mode: print(f"{LogColors.ERROR}[ERROR]{LogColors.RESET} {message}") elif level == "debug": # Hinzugefügt für detaillierteres Debugging logging.debug(message) if debug_mode: print(f"{LogColors.DEBUG}[DEBUG]{LogColors.RESET} {message}") # --- Hilfsfunktionen zur Prüfung der Tool-Verfügbarkeit --- def check_command_available(cmd): """Prüft, ob ein Systembefehl im PATH verfügbar ist.""" return subprocess.run(['which', cmd], capture_output=True, text=True).returncode == 0 NOTIFY_SEND_AVAILABLE = False SOUND_PLAYER = None SOUND_FILE = None def setup_tool_availability(debug_mode=False): """Richtet die globalen Variablen für die Tool-Verfügbarkeit ein.""" global NOTIFY_SEND_AVAILABLE, SOUND_PLAYER, SOUND_FILE # Prüfe notify-send if check_command_available('notify-send'): NOTIFY_SEND_AVAILABLE = True log_message("info", "notify-send gefunden.", debug_mode) else: log_message("warning", "notify-send nicht gefunden. Desktop-Benachrichtigungen sind deaktiviert. Bitte libnotify-bin installieren.", debug_mode) NOTIFY_SEND_AVAILABLE = False # Prüfe Sound-Tools if check_command_available('paplay'): SOUND_PLAYER = "paplay" SOUND_FILE = "/usr/share/sounds/freedesktop/stereo/dialog-warning.oga" log_message("info", f"Sound-Player gefunden: {SOUND_PLAYER}", debug_mode) elif check_command_available('aplay'): SOUND_PLAYER = "aplay" SOUND_FILE = "/usr/share/sounds/alsa/Front_Center.wav" log_message("info", f"Sound-Player gefunden: {SOUND_PLAYER}", debug_mode) elif check_command_available('canberra-gtk-play'): SOUND_PLAYER = "canberra-gtk-play" SOUND_FILE = "dialog-warning" # canberra-gtk-play nutzt Namen, keine Pfade log_message("info", f"Sound-Player gefunden: {SOUND_PLAYER}", debug_mode) else: log_message("warning", "Kein passender Sound-Player gefunden. Sounds sind deaktiviert.", debug_mode) SOUND_PLAYER = None # --- Funktionen für Aktionen --- def send_notify(username, message, debug_mode=False): """Sendet eine Desktop-Benachrichtigung an den angegebenen Benutzer.""" if not NOTIFY_SEND_AVAILABLE: log_message("info", f"Benachrichtigung übersprungen (notify-send nicht verfügbar): {message}", debug_mode) return try: user_id = os.getuid() # Get the UID of the current user running the script if os.name == 'posix': # Linux/macOS # Try to get the session bus address from /run/user//bus # This is tricky as we need the target user's UID, not the current script's UID # If the script runs as root, os.getuid() will return 0. # We need to get the actual user's UID for the DBUS_SESSION_BUS_ADDRESS. # This is complex and might require 'loginctl show-user ' or parsing 'ps -eo user,args' # For simplicity, we'll try to guess the user's UID or rely on environment if run as that user # For now, let's assume the user's UID is known or can be found. # A more robust way is to get it from 'id -u ' try: target_uid = subprocess.check_output(['id', '-u', username], text=True).strip() except subprocess.CalledProcessError: log_message("error", f"Konnte UID für Benutzer '{username}' nicht ermitteln.", debug_mode) return dbus_address_path = f"/run/user/{target_uid}/bus" # Check if the D-Bus session bus address file exists and is accessible if not os.path.exists(dbus_address_path): log_message("warning", f"D-Bus Session Bus Datei für Benutzer {username} nicht gefunden: {dbus_address_path}. Benachrichtigung könnte fehlschlagen.", debug_mode) # Attempt to find it via environment of user's processes (more complex) # For now, we proceed, but this is a common point of failure. # We need to ensure the command is run in the context of the user's display/session # This typically requires running as the user with their DISPLAY/DBUS_SESSION_BUS_ADDRESS set # which is hard when the Python script itself is run as root. # The shell script uses 'sudo -u "$USER_TO_MANAGE" DISPLAY=:0 DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/$(id -u $USER_TO_MANAGE)/bus notify-send' # We mimic this: cmd = ['sudo', '-u', username, 'DISPLAY=:0', f'DBUS_SESSION_BUS_ADDRESS=unix:path={dbus_address_path}', 'notify-send', "Safe Kiddo", message] log_message("debug", f"notify-send Befehl: {' '.join(cmd)}", debug_mode) result = subprocess.run(cmd, capture_output=True, text=True, check=False) if result.returncode != 0: log_message("error", f"Fehler beim Senden der Benachrichtigung an {username}: {result.stderr}", debug_mode) else: log_message("info", f"Benachrichtigung an {username} gesendet: '{message}'", debug_mode) elif os.name == 'nt': # Windows - using win10toast or similar log_message("warning", "Desktop-Benachrichtigungen für Windows noch nicht implementiert.", debug_mode) # You would use a library like 'win10toast' here: # from win10toast import ToastNotifier # ToastNotifier().show_toast("Safe Kiddo", message, duration=5) else: log_message("warning", f"Desktop-Benachrichtigungen für OS '{os.name}' nicht unterstützt.", debug_mode) except Exception as e: log_message("error", f"Unerwarteter Fehler beim Senden der Benachrichtigung: {e}", debug_mode) def play_beep_sound(debug_mode=False): """Spielt einen Warnton ab, wenn ein Sound-Player verfügbar ist.""" if not SOUND_PLAYER: log_message("info", "Sound übersprungen (kein Player verfügbar).", debug_mode) return try: if SOUND_PLAYER == "canberra-gtk-play": cmd = [SOUND_PLAYER, '-i', SOUND_FILE, '-d', "SafeKiddo"] else: cmd = [SOUND_PLAYER, SOUND_FILE] log_message("debug", f"Sound-Befehl: {' '.join(cmd)}", debug_mode) # subprocess.Popen, damit es im Hintergrund läuft und das Skript nicht blockiert subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) log_message("info", "Sound abgespielt.", debug_mode) except Exception as e: log_message("error", f"Fehler beim Abspielen des Sounds: {e}", debug_mode) def is_user_logged_in(username, debug_mode=False): """Prüft, ob der angegebene Benutzer aktuell eingeloggt ist.""" log_message("info", f"Prüfe, ob Benutzer '{username}' eingeloggt ist...", debug_mode) for user in psutil.users(): if user.name == username: log_message("info", f"Benutzer '{username}' ist eingeloggt.", debug_mode) return True log_message("info", f"Benutzer '{username}' ist NICHT eingeloggt.", debug_mode) return False def disable_user(username, countdown_mode, sound_mode, countdown_time, debug_mode=False): """Deaktiviert ein Benutzerkonto und führt bei aktivem Login Countdown und Shutdown durch.""" log_message("info", f"Deaktiviere Benutzer '{username}'...", debug_mode) try: # Benutzerkonto sperren # usermod -L ist ein Linux-Befehl. Für Windows/macOS wären andere Befehle nötig. # Hier muss das Skript mit Root-Rechten ausgeführt werden (z.B. sudo python3 user_control_agent.py) result = subprocess.run(['sudo', 'usermod', '-L', username], capture_output=True, text=True, check=True) log_message("info", f"Benutzer '{username}' Konto gesperrt: {result.stdout.strip()}", debug_mode) except subprocess.CalledProcessError as e: log_message("error", f"Fehler beim Sperren des Benutzers '{username}': {e.stderr.strip()}", debug_mode) return False # Abbruch bei Fehler user_was_logged_in = is_user_logged_in(username, debug_mode) if user_was_logged_in: log_message("info", f"Benutzer '{username}' ist eingeloggt. Starte Countdown/Shutdown-Prozess...", debug_mode) if countdown_mode: log_message("info", f"Starte dramatischen Countdown über {countdown_time} Sekunden...", debug_mode) for i in range(countdown_time, 0, -1): send_notify(username, f"ACHTUNG! Shutdown in {i} Sekunden!", debug_mode) if sound_mode: play_beep_sound(debug_mode) time.sleep(1) else: log_message("info", "Sende Warnung (ohne Countdown)...", debug_mode) send_notify(username, f"Shutdown in {countdown_time} Sekunden! Speicher deine Arbeit/Spiel!", debug_mode) log_message("info", f"Warten für {countdown_time} Sekunden...", debug_mode) time.sleep(countdown_time) log_message("info", f"Benutzer '{username}' abmelden (falls noch eingeloggt)...", debug_mode) try: # Beende alle Prozesse des Benutzers. psutil ist hier robuster. for proc in psutil.process_iter(['pid', 'name', 'username']): if proc.info['username'] == username: try: proc.kill() log_message("debug", f"Prozess {proc.info['name']} (PID: {proc.info['pid']}) beendet.", debug_mode) except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess): log_message("debug", f"Konnte Prozess {proc.info['name']} (PID: {proc.info['pid']}) nicht beenden (bereits beendet/Zugriff verweigert).", debug_mode) # Alternative (weniger robust, da shell-Befehl): subprocess.run(['sudo', 'pkill', '-KILL', '-u', username], check=False) log_message("info", f"Alle Prozesse von Benutzer '{username}' beendet.", debug_mode) except Exception as e: log_message("error", f"Fehler beim Beenden der Prozesse für Benutzer '{username}': {e}", debug_mode) log_message("info", "Rechner wird jetzt heruntergefahren...", debug_mode) try: # Shutdown-Befehl - erfordert ebenfalls Root-Rechte subprocess.run(['sudo', 'shutdown', 'now'], check=True) log_message("info", "Shutdown-Befehl gesendet.", debug_mode) except subprocess.CalledProcessError as e: log_message("error", f"Fehler beim Senden des Shutdown-Befehls: {e.stderr.strip()}", debug_mode) return False # Fehler beim Shutdown else: log_message("info", f"Benutzer '{username}' ist NICHT eingeloggt. Countdown und Shutdown werden übersprungen.", debug_mode) log_message("info", f"Benutzer '{username}' wurde deaktiviert.", debug_mode) return True # Aktion erfolgreich (unabhängig vom Shutdown) def enable_user(username, debug_mode=False): """Aktiviert ein Benutzerkonto.""" log_message("info", f"Aktiviere Benutzer '{username}'...", debug_mode) try: result = subprocess.run(['sudo', 'usermod', '-U', username], capture_output=True, text=True, check=True) log_message("info", f"Benutzer '{username}' Konto entsperrt: {result.stdout.strip()}", debug_mode) return True except subprocess.CalledProcessError as e: log_message("error", f"Fehler beim Entsperren des Benutzers '{username}': {e.stderr.strip()}", debug_mode) return False # --- Haupt-Logik bei direktem Ausführen (CLI) --- if __name__ == "__main__": parser = argparse.ArgumentParser(description="Deaktiviert/Aktiviert Benutzerkonten mit optionalem Countdown und Shutdown.") parser.add_argument("username", help="Der Benutzername, der verwaltet werden soll.") parser.add_argument("action", choices=["disable", "enable"], help="Die Aktion: 'disable' oder 'enable'.") parser.add_argument("--countdown", action="store_true", help="Führt einen Countdown mit Benachrichtigungen vor dem Shutdown durch.") parser.add_argument("--sound", action="store_true", help="Spielt einen Warnton während des Countdowns ab.") parser.add_argument("--time", type=int, default=60, help="Dauer des Countdowns in Sekunden (Standard: 60).") parser.add_argument("--debug", action="store_true", help="Aktiviert detailliertes Debug-Logging auf der CLI.") args = parser.parse_args() # Debug-Modus für Log-Nachrichten setzen set_debug_mode(args.debug) # Verfügbarkeit von System-Tools prüfen setup_tool_availability(args.debug) if args.action == "disable": log_message("info", f"Beginne 'disable' Aktion für Benutzer '{args.username}'.", args.debug) disable_user(args.username, args.countdown, args.sound, args.time, args.debug) elif args.action == "enable": log_message("info", f"Beginne 'enable' Aktion für Benutzer '{args.username}'.", args.debug) enable_user(args.username, args.debug) log_message("info", f"Aktion für Benutzer '{args.username}' abgeschlossen.", args.debug)