first commit
This commit is contained in:
stephan
275
user_control_agent.py
Normal file
275
user_control_agent.py
Normal file
@ -0,0 +1,275 @@
|
||||
# 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/<UID>/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 <username>' 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 <username>'
|
||||
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)
|
||||
Reference in New Issue
Block a user