diff --git a/README.md b/README.md index 53627b8..914c327 100644 --- a/README.md +++ b/README.md @@ -4,13 +4,32 @@ Werkzeuge für Kerio Connect auf der Kommandozeile. | Datei | Zweck | | --- | --- | +| [`kerio_tui.py`](kerio_tui.py) | Assistent mit Menüführung — ohne Kommandozeilenparameter | | [`kerio_rename_user.py`](kerio_rename_user.py) | Benutzer umbenennen — Login-Name **und** Mailbox-Verzeichnis im Store | | [`kerio_users_config.py`](kerio_users_config.py) | Lesen und Ändern von `users.cfg` | | [`kerio_common.py`](kerio_common.py) | Gemeinsame Bausteine: Admin-API-Client, Dienststeuerung | -Das Script läuft **lokal auf dem Kerio-Server als root**, braucht nur die +Alles läuft **lokal auf dem Kerio-Server als root**, braucht nur die Python-Standardbibliothek und bietet `--dry-run`. +## Der Assistent + +Wer nicht mit Parametern hantieren will: + +```bash +sudo ./kerio_tui.py +``` + +Der Assistent zeigt die vorhandenen Benutzer zur Auswahl, findet **verwaiste +Mailbox-Verzeichnisse von selbst** (Verzeichnisse ohne zugehörigen Benutzer), +fragt die nötigen Angaben ab und zeigt vor dem Ausführen den gleichwertigen +Kommandozeilenaufruf — wer ihn kennt, tippt ihn beim nächsten Mal direkt. + +Bedienung: Pfeiltasten oder `j`/`k`, Enter bestätigt, `q` oder Esc geht zurück. + +Geändert wird erst nach einer ausdrücklichen Bestätigung; bis dahin lässt sich +jeder Schritt abbrechen. + ## Getestet Gegen **Kerio Connect 10.0.9 patch 2** auf Debian 12, mit lizenziertem Server @@ -39,6 +58,13 @@ Und für `--adopt`, mit gelöschtem Altbenutzer und neu angelegtem Zielbenutzer: | Neue Mail an den Zielbenutzer | zugestellt | | Verwaistes Verzeichnis | keines übrig | +Der Assistent wurde über ein PTY durch alle Abläufe gefahren; dabei kam heraus, +dass `keypad(True)` auf diesem Server die Pfeiltasten **nicht** auflöst — +`getch()` liefert `27, 91, 66` statt `KEY_DOWN`. Ein fehlendes oder +unpassendes terminfo reicht dafür. Die Oberfläche löst die Sequenzen deshalb +selbst auf, sonst würde das führende ESC als „zurück" gelten und der Assistent +bei jedem Pfeildruck herausspringen. + Geändert haben sich nur Kerios eigene Index- und Statusdateien (`.journal.db`, `status.fld`, `index.fld`) — die schreibt der Server bei jedem Zugriff neu. @@ -48,6 +74,7 @@ Dazu Unit-Tests unter [tests/](tests/): python3 tests/test_store.py python3 tests/test_users_config.py python3 tests/test_adopt.py +python3 tests/test_tui.py ``` --- @@ -228,6 +255,10 @@ legt anschließend in der Administrationskonsole einen Alias an. - **Vorher ein Backup des Stores anlegen.** Das Script verschiebt Verzeichnisse. - Erst an einem Testbenutzer ausprobieren. - Der Benutzer sollte während des Laufs nicht angemeldet sein. +- Bricht die SSH-Verbindung weg, während Kerio gestoppt ist, fängt das Script + das Signal ab und **startet den Dienst wieder**, bevor es sich beendet. + Verlassen sollte man sich darauf trotzdem nicht — besser in `tmux` oder + `screen` arbeiten. - Nach der Umbenennung müssen sich alle Mail-Clients mit dem neuen Login neu anmelden. - Gruppenmitgliedschaften bleiben erhalten: Kerio referenziert Gruppen über diff --git a/kerio_common.py b/kerio_common.py index 32731ef..8227613 100644 --- a/kerio_common.py +++ b/kerio_common.py @@ -8,9 +8,11 @@ Gemeinsame Bausteine fuer die Kerio-Connect-Scripts: from __future__ import annotations +import contextlib import http.cookiejar import json import shutil +import signal import ssl import subprocess import sys @@ -282,6 +284,37 @@ class ServiceController: f"Dienst '{self.name}' laeuft nach {self.wait_seconds}s immer noch. Abbruch." ) + @contextlib.contextmanager + def abgesichert(self, war_aktiv: bool): + """Startet den Dienst wieder, wenn das Script abgeschossen wird. + + Diese Scripts laufen ueblicherweise ueber SSH. Bricht die Verbindung + weg, kommt SIGHUP - und ohne Vorkehrung bliebe Kerio gestoppt + zurueck, weil der Prozess stirbt, bevor er ihn wieder starten kann. + Dasselbe gilt fuer SIGTERM und Strg-C. + """ + def behandle(signalnummer, _rahmen): + print(f"\n Signal {signalnummer} empfangen - Kerio wird wieder " + "gestartet, dann Abbruch.", file=sys.stderr) + if war_aktiv: + self.start() + raise SystemExit(130) + + vorher = {} + for signalnummer in (signal.SIGHUP, signal.SIGTERM, signal.SIGINT): + try: + vorher[signalnummer] = signal.signal(signalnummer, behandle) + except (ValueError, OSError): + pass # z. B. wenn nicht im Hauptthread + try: + yield + finally: + for signalnummer, handler in vorher.items(): + try: + signal.signal(signalnummer, handler) + except (ValueError, OSError): + pass + def start(self) -> None: if self.mode == "none": return diff --git a/kerio_rename_user.py b/kerio_rename_user.py index c73ff0b..ce0fe66 100755 --- a/kerio_rename_user.py +++ b/kerio_rename_user.py @@ -373,35 +373,39 @@ def adopt_store(args: argparse.Namespace, config: UsersConfig, print() was_running = service.is_running() - service.stop() done: list[tuple[Path, Path]] = [] - try: - check_rename_targets(moves) - done = apply_store_renames(moves) + # Ab hier ist Kerio gestoppt. Der Waechter sorgt dafuer, dass der Dienst + # auch dann wieder hochkommt, wenn das Script abgeschossen wird. + with service.abgesichert(bool(was_running)): + service.stop() + try: + check_rename_targets(moves) + done = apply_store_renames(moves) - # Die uebernommene Mailbox traegt noch die Identitaet des alten - # Benutzers. stats.usr legt Kerio selbst neu an, die Kontaktkarte - # nicht - die muss umgeschrieben werden. - for _, target in done: - if ".abgeloest-" in target.name: - continue - if rewrite_personal_vcard( - target, - f"{args.old_login}@{domain}", - f"{args.new_login}@{domain}", - target_user["fullName"]): - print(f" Kontaktkarte auf {target_user['fullName']} umgeschrieben") - stats = target / "stats.usr" - if stats.is_file(): - stats.rename(target / f"stats.usr.alt-{stamp}") - print(" Beiseite: stats.usr (Kerio legt sie neu an)") - except (AbortedError, OSError) as exc: - print(f"\nFEHLER: {exc}", file=sys.stderr) - revert_store_renames(done) - if was_running: - service.start() - return 2 + # Die uebernommene Mailbox traegt noch die Identitaet des alten + # Benutzers. stats.usr legt Kerio selbst neu an, die Kontaktkarte + # nicht - die muss umgeschrieben werden. + for _, target in done: + if ".abgeloest-" in target.name: + continue + if rewrite_personal_vcard( + target, + f"{args.old_login}@{domain}", + f"{args.new_login}@{domain}", + target_user["fullName"]): + print(f" Kontaktkarte auf {target_user['fullName']} " + "umgeschrieben") + stats = target / "stats.usr" + if stats.is_file(): + stats.rename(target / f"stats.usr.alt-{stamp}") + print(" Beiseite: stats.usr (Kerio legt sie neu an)") + except (AbortedError, OSError) as exc: + print(f"\nFEHLER: {exc}", file=sys.stderr) + revert_store_renames(done) + if was_running: + service.start() + return 2 if was_running: service.start() @@ -513,54 +517,58 @@ def rename_user(args: argparse.Namespace, config: UsersConfig, # -- Schreiben: users.cfg und Verzeichnisse in einem Stopp-Fenster ---- print() was_running = service.is_running() - service.stop() done: list[tuple[Path, Path]] = [] backup: Path | None = None - try: - # Nach dem Stoppen neu einlesen: Kerio hat users.cfg beim Beenden - # aus dem Speicher zurueckgeschrieben, unsere Kopie ist veraltet. - fresh = UsersConfig(config.path) - backup = fresh.backup(BACKUP_SUFFIX) - print(f" Backup: {backup}") + # Ab hier ist Kerio gestoppt. Der Waechter sorgt dafuer, dass der Dienst + # auch dann wieder hochkommt, wenn das Script abgeschossen wird - etwa + # weil die SSH-Verbindung wegbricht. + with service.abgesichert(bool(was_running)): + service.stop() + try: + # Nach dem Stoppen neu einlesen: Kerio hat users.cfg beim Beenden + # aus dem Speicher zurueckgeschrieben, unsere Kopie ist veraltet. + fresh = UsersConfig(config.path) + backup = fresh.backup(BACKUP_SUFFIX) + print(f" Backup: {backup}") - changed = fresh.rename_user( - old_login, new_login, domain, - args.full_name if needs_fullname else None) - if changed == 0: - raise AbortedError( - "Kein passender Eintrag in users.cfg gefunden - nichts geaendert." - ) - fresh.save() - print(f" users.cfg: {changed} Eintraege angepasst " - f"(Listen: User, UserAdditionalData)") + changed = fresh.rename_user( + old_login, new_login, domain, + args.full_name if needs_fullname else None) + if changed == 0: + raise AbortedError( + "Kein passender Eintrag in users.cfg gefunden - nichts geaendert." + ) + fresh.save() + print(f" users.cfg: {changed} Eintraege angepasst " + f"(Listen: User, UserAdditionalData)") - if renames: - # Zielpruefung wiederholen: der Zustand kann sich seit der - # Vorpruefung geaendert haben. - check_rename_targets(renames) - done = apply_store_renames(renames) + if renames: + # Zielpruefung wiederholen: der Zustand kann sich seit der + # Vorpruefung geaendert haben. + check_rename_targets(renames) + done = apply_store_renames(renames) - # Die Kontaktkarte traegt noch die alte Adresse - Kerio zieht dort - # nur den Namen nach, nicht die Adresse. - display_name = args.full_name if needs_fullname else user["fullName"] - for _, target in done: - if rewrite_personal_vcard(target, f"{old_login}@{domain}", - f"{new_login}@{domain}", display_name): - print(" Kontaktkarte auf die neue Adresse umgeschrieben") - except (AbortedError, OSError) as exc: - print(f"\nFEHLER: {exc}", file=sys.stderr) - revert_store_renames(done) - if backup and backup.is_file(): - try: - shutil.copy2(backup, config.path) - print(f" users.cfg aus {backup} zurueckgesetzt.", file=sys.stderr) - except OSError as restore_exc: - print(f" WARNUNG: Zuruecksetzen von users.cfg fehlgeschlagen: " - f"{restore_exc}", file=sys.stderr) - if was_running: - service.start() - return 2 + # Die Kontaktkarte traegt noch die alte Adresse - Kerio zieht dort + # nur den Namen nach, nicht die Adresse. + display_name = args.full_name if needs_fullname else user["fullName"] + for _, target in done: + if rewrite_personal_vcard(target, f"{old_login}@{domain}", + f"{new_login}@{domain}", display_name): + print(" Kontaktkarte auf die neue Adresse umgeschrieben") + except (AbortedError, OSError) as exc: + print(f"\nFEHLER: {exc}", file=sys.stderr) + revert_store_renames(done) + if backup and backup.is_file(): + try: + shutil.copy2(backup, config.path) + print(f" users.cfg aus {backup} zurueckgesetzt.", file=sys.stderr) + except OSError as restore_exc: + print(f" WARNUNG: Zuruecksetzen von users.cfg fehlgeschlagen: " + f"{restore_exc}", file=sys.stderr) + if was_running: + service.start() + return 2 if was_running: service.start() diff --git a/kerio_tui.py b/kerio_tui.py new file mode 100644 index 0000000..78e69a1 --- /dev/null +++ b/kerio_tui.py @@ -0,0 +1,545 @@ +#!/usr/bin/env python3 +""" +Bedienoberflaeche fuer die Kerio-Connect-Werkzeuge - fuer alle, die nicht mit +Kommandozeilenparametern hantieren moechten. + +Der Assistent fragt Schritt fuer Schritt ab, was zu tun ist, zeigt am Ende den +gleichwertigen Kommandozeilenaufruf und fuehrt ihn dann aus. Wer den Befehl +kennt, kann ihn beim naechsten Mal direkt tippen. + +Bedienung: Pfeiltasten zum Auswaehlen, Enter zum Bestaetigen, q oder Esc zurueck. + +Aufruf ohne Parameter: + + sudo ./kerio_tui.py + +Fuer den Trockenlauf reichen normale Rechte - geaendert wird erst nach einer +ausdruecklichen Bestaetigung. +""" + +from __future__ import annotations + +import curses +import os +import sys +from pathlib import Path + +from kerio_common import ( + DEFAULT_INSTALL_DIR, + DEFAULT_STORE_DIR, + AbortedError, +) +from kerio_users_config import UsersConfig, validate_login +import kerio_rename_user as rename_tool + +TITLE = "Kerio Connect - Benutzerverwaltung" + + +class Abbruch(Exception): + """Der Benutzer hat den Assistenten verlassen.""" + + +# ========================================================================== +# Bausteine der Oberflaeche +# ========================================================================== + + +def _setup_colors() -> dict[str, int]: + """Farbpaare anlegen, wenn das Terminal sie kann - sonst neutrale Werte.""" + styles = {"titel": curses.A_BOLD, "hinweis": curses.A_DIM, + "auswahl": curses.A_REVERSE, "warnung": curses.A_BOLD, + "gut": curses.A_BOLD} + if not curses.has_colors(): + return styles + curses.start_color() + try: + curses.use_default_colors() + hintergrund = -1 + except curses.error: + hintergrund = curses.COLOR_BLACK + curses.init_pair(1, curses.COLOR_CYAN, hintergrund) + curses.init_pair(2, curses.COLOR_YELLOW, hintergrund) + curses.init_pair(3, curses.COLOR_GREEN, hintergrund) + curses.init_pair(4, curses.COLOR_RED, hintergrund) + styles["titel"] = curses.color_pair(1) | curses.A_BOLD + styles["hinweis"] = curses.color_pair(2) + styles["gut"] = curses.color_pair(3) | curses.A_BOLD + styles["warnung"] = curses.color_pair(4) | curses.A_BOLD + return styles + + +def _schreibe(win, y: int, x: int, text: str, attr: int = 0) -> None: + """Schreibt Text und schneidet ab, statt bei schmalem Terminal zu scheitern.""" + hoehe, breite = win.getmaxyx() + if y < 0 or y >= hoehe or x >= breite: + return + try: + win.addnstr(y, x, text, max(0, breite - x - 1), attr) + except curses.error: + pass + + +def _taste(win) -> int: + """Liest eine Taste und loest rohe Escape-Sequenzen selbst auf. + + Eigentlich soll keypad(True) Pfeiltasten in KEY_UP/KEY_DOWN uebersetzen. + Wo die terminfo-Datenbank fehlt oder nicht zum gesetzten TERM passt - + auf frisch installierten Servern keine Seltenheit - liefert getch() + stattdessen ESC, '[' und 'B' einzeln. Das fuehrende ESC gaelte dann als + "zurueck", und die Oberflaeche spraenge bei jedem Pfeildruck heraus. + """ + taste = win.getch() + if taste != 27: + return taste + + # Nach ESC nachsehen, ob eine Sequenz folgt - ohne zu blockieren, sonst + # haengt ein einzelner ESC-Druck. + win.nodelay(True) + try: + zweite = -1 + for _ in range(4): + zweite = win.getch() + if zweite != -1: + break + curses.napms(5) + if zweite == -1: + return 27 # wirklich nur ESC + if zweite not in (ord("["), ord("O")): + return 27 + dritte = -1 + for _ in range(4): + dritte = win.getch() + if dritte != -1: + break + curses.napms(5) + return { + ord("A"): curses.KEY_UP, + ord("B"): curses.KEY_DOWN, + ord("C"): curses.KEY_RIGHT, + ord("D"): curses.KEY_LEFT, + ord("H"): curses.KEY_HOME, + ord("F"): curses.KEY_END, + }.get(dritte, -1) + finally: + win.nodelay(False) + + +def _rahmen(win, styles: dict[str, int], untertitel: str = "") -> int: + """Zeichnet Kopfzeile und gibt die erste freie Zeile zurueck.""" + win.erase() + _schreibe(win, 0, 2, TITLE, styles["titel"]) + if untertitel: + _schreibe(win, 1, 2, untertitel, styles["hinweis"]) + return 3 + return 2 + + +def menue(win, styles: dict[str, int], untertitel: str, eintraege: list[str], + fusszeile: str = "Pfeiltasten Enter=Auswahl q=zurueck") -> int: + """Laesst einen Eintrag auswaehlen und gibt seinen Index zurueck.""" + if not eintraege: + raise Abbruch + pos = 0 + while True: + oben = _rahmen(win, styles, untertitel) + hoehe, _ = win.getmaxyx() + platz = max(1, hoehe - oben - 2) + anfang = max(0, min(pos - platz // 2, len(eintraege) - platz)) + + for i, eintrag in enumerate(eintraege[anfang:anfang + platz]): + echt = anfang + i + marke = "> " if echt == pos else " " + attr = styles["auswahl"] if echt == pos else 0 + _schreibe(win, oben + i, 2, f"{marke}{eintrag}", attr) + + if len(eintraege) > platz: + _schreibe(win, hoehe - 2, 2, + f"({pos + 1}/{len(eintraege)})", styles["hinweis"]) + _schreibe(win, hoehe - 1, 2, fusszeile, styles["hinweis"]) + win.refresh() + + taste = _taste(win) + if taste in (curses.KEY_UP, ord("k")): + pos = (pos - 1) % len(eintraege) + elif taste in (curses.KEY_DOWN, ord("j")): + pos = (pos + 1) % len(eintraege) + elif taste == curses.KEY_HOME: + pos = 0 + elif taste == curses.KEY_END: + pos = len(eintraege) - 1 + elif taste in (curses.KEY_ENTER, 10, 13): + return pos + elif taste in (ord("q"), 27): + raise Abbruch + + +def eingabe(win, styles: dict[str, int], untertitel: str, beschriftung: str, + vorgabe: str = "", pflicht: bool = True) -> str: + """Fragt eine Zeile Text ab. Leere Eingabe bedeutet: Vorgabe uebernehmen.""" + puffer = vorgabe + fehler = "" + while True: + oben = _rahmen(win, styles, untertitel) + _schreibe(win, oben, 2, beschriftung) + _schreibe(win, oben + 2, 4, puffer + "_", styles["auswahl"]) + if vorgabe: + _schreibe(win, oben + 4, 2, f"Vorschlag: {vorgabe}", styles["hinweis"]) + if fehler: + _schreibe(win, oben + 5, 2, fehler, styles["warnung"]) + hoehe, _ = win.getmaxyx() + _schreibe(win, hoehe - 1, 2, + "Enter=uebernehmen Esc=zurueck Rueckschritt=loeschen", + styles["hinweis"]) + win.refresh() + + taste = _taste(win) + if taste in (curses.KEY_ENTER, 10, 13): + wert = puffer.strip() + if not wert and pflicht: + fehler = "Bitte etwas eingeben." + continue + return wert + if taste == 27: + raise Abbruch + if taste in (curses.KEY_BACKSPACE, 127, 8): + puffer = puffer[:-1] + fehler = "" + elif 32 <= taste < 127: + puffer += chr(taste) + fehler = "" + + +def zeige_text(win, styles: dict[str, int], untertitel: str, + zeilen: list[tuple[str, str]], + fusszeile: str = "Beliebige Taste") -> None: + """Zeigt Textzeilen als (stil, text)-Paare.""" + oben = _rahmen(win, styles, untertitel) + for i, (stil, text) in enumerate(zeilen): + _schreibe(win, oben + i, 2, text, styles.get(stil, 0)) + hoehe, _ = win.getmaxyx() + _schreibe(win, hoehe - 1, 2, fusszeile, styles["hinweis"]) + win.refresh() + _taste(win) + + +def bestaetige(win, styles: dict[str, int], untertitel: str, + zeilen: list[tuple[str, str]]) -> bool: + oben = _rahmen(win, styles, untertitel) + for i, (stil, text) in enumerate(zeilen): + _schreibe(win, oben + i, 2, text, styles.get(stil, 0)) + hoehe, _ = win.getmaxyx() + _schreibe(win, hoehe - 2, 2, "Wirklich durchfuehren?", styles["warnung"]) + _schreibe(win, hoehe - 1, 2, "j = ja, alles andere = abbrechen", + styles["hinweis"]) + win.refresh() + return _taste(win) in (ord("j"), ord("J"), ord("y"), ord("Y")) + + +# ========================================================================== +# Fachliches +# ========================================================================== + + +def verwaiste_verzeichnisse(store_dir: Path, domain: str, + bekannte_logins: set[str]) -> list[str]: + """Verzeichnisse im Store, zu denen es keinen Benutzer (mehr) gibt. + + Genau die Kandidaten fuer die Uebernahme. Kerios eigene Ordner wie + '#public' und bereits abgeloeste Sicherungen bleiben aussen vor. + """ + # Nicht darauf verlassen, dass der Aufrufer schon kleingeschrieben hat - + # ein uebersehener Benutzer wuerde hier als verwaist erscheinen. + bekannte = {name.lower() for name in bekannte_logins} + gefunden: list[str] = [] + for unterbaum in rename_tool.STORE_SUBTREES: + domain_dir = rename_tool.find_dir_ci(store_dir / unterbaum, domain) + if domain_dir is None or not domain_dir.is_dir(): + continue + for kind in sorted(domain_dir.iterdir()): + if not kind.is_dir() or kind.name.startswith("#"): + continue + if ".abgeloest-" in kind.name: + continue + if kind.name.lower() in bekannte: + continue + if kind.name not in gefunden: + gefunden.append(kind.name) + return gefunden + + +def _befehl_text(argumente: list[str]) -> str: + return "sudo ./kerio_rename_user.py " + " ".join( + f'"{a}"' if " " in a else a for a in argumente) + + +def ablauf_umbenennen(win, styles, config: UsersConfig, + store_dir: Path) -> list[str] | None: + """Sammelt die Angaben fuer eine Umbenennung.""" + domains = config.domains() + if not domains: + zeige_text(win, styles, "Umbenennen", + [("warnung", "In users.cfg steht keine Domain.")]) + return None + domain = domains[menue(win, styles, "Umbenennen - Domain waehlen", domains)] + + benutzer = config.users(domain) + if not benutzer: + zeige_text(win, styles, "Umbenennen", + [("warnung", f"Keine Benutzer in '{domain}'.")]) + return None + + beschriftungen = [] + for b in benutzer: + zusatz = "" if config.is_internal(b["loginName"], domain) \ + else " [Verzeichnisdienst]" + beschriftungen.append(f"{b['loginName']:<26} {b['fullName']}{zusatz}") + gewaehlt = benutzer[menue(win, styles, + f"Umbenennen - Benutzer in {domain}", + beschriftungen)] + alt = gewaehlt["loginName"] + + if not config.is_internal(alt, domain): + zeige_text(win, styles, "Umbenennen", [ + ("warnung", f"'{alt}' stammt aus einem Verzeichnisdienst."), + ("", "Der Login-Name muss dort geaendert werden, nicht in Kerio."), + ]) + return None + + while True: + neu = eingabe(win, styles, f"Neuer Login-Name fuer {alt}", + "Neuer Login-Name (ohne @domain):", alt) + if neu == alt: + break + try: + validate_login(neu) + except AbortedError as exc: + zeige_text(win, styles, "Ungueltig", [("warnung", str(exc))]) + continue + if config.find_user(neu, domain) is not None: + zeige_text(win, styles, "Schon vergeben", + [("warnung", f"'{neu}' gibt es in {domain} bereits.")]) + continue + break + + voll = eingabe(win, styles, f"Angezeigter Name fuer {neu}", + "Angezeigter Name (leer = unveraendert lassen):", + gewaehlt["fullName"], pflicht=False) + + if neu == alt and voll == gewaehlt["fullName"]: + zeige_text(win, styles, "Nichts zu tun", + [("", "Weder Login-Name noch angezeigter Name geaendert.")]) + return None + + argumente = ["--domain", domain, "--old-login", alt] + if neu != alt: + argumente += ["--new-login", neu] + if voll and voll != gewaehlt["fullName"]: + argumente += ["--full-name", voll] + + zeilen = [ + ("", f"Domain : {domain}"), + ("", f"Login-Name : {alt} -> {neu}"), + ("", f"Angezeigter Name: {gewaehlt['fullName']} -> {voll or gewaehlt['fullName']}"), + ("", ""), + ("hinweis", "Der Kerio-Dienst wird dabei kurz gestoppt."), + ("", ""), + ("", "Gleichwertiger Aufruf:"), + ("hinweis", _befehl_text(argumente)), + ] + if neu != alt: + weitere = config.other_occurrences(alt, domain) + if weitere: + zeilen.insert(3, ("warnung", + f"Achtung: '{alt}' kommt auch vor in: " + + ", ".join(weitere))) + return argumente if bestaetige(win, styles, "Zusammenfassung", zeilen) else None + + +def ablauf_uebernehmen(win, styles, config: UsersConfig, + store_dir: Path) -> list[str] | None: + """Sammelt die Angaben fuer die Uebernahme eines verwaisten Verzeichnisses.""" + domains = config.domains() + if not domains: + zeige_text(win, styles, "Uebernehmen", + [("warnung", "In users.cfg steht keine Domain.")]) + return None + domain = domains[menue(win, styles, "Uebernehmen - Domain waehlen", domains)] + + bekannt = {b["loginName"].lower() for b in config.users(domain)} + verwaist = verwaiste_verzeichnisse(store_dir, domain, bekannt) + if not verwaist: + zeige_text(win, styles, "Uebernehmen", [ + ("gut", f"In '{domain}' gibt es kein verwaistes Verzeichnis."), + ("", ""), + ("", "Zu jedem Mailbox-Verzeichnis existiert ein Benutzer."), + ]) + return None + + beschriftungen = [] + for name in verwaist: + pfad = rename_tool.find_dir_ci( + rename_tool.find_dir_ci(store_dir / "mail", domain) or store_dir, name) + anzahl = rename_tool.count_mails(pfad) if pfad else 0 + beschriftungen.append(f"{name:<26} {anzahl} Mail(s)") + quelle = verwaist[menue(win, styles, + f"Verwaistes Verzeichnis in {domain}", + beschriftungen)] + + benutzer = config.users(domain) + if not benutzer: + zeige_text(win, styles, "Uebernehmen", + [("warnung", f"Keine Benutzer in '{domain}'.")]) + return None + ziel_beschriftung = [f"{b['loginName']:<26} {b['fullName']}" for b in benutzer] + ziel = benutzer[menue(win, styles, + f"Wer soll '{quelle}' bekommen?", + ziel_beschriftung)]["loginName"] + + store_pfad = rename_tool.find_dir_ci(store_dir / "mail", domain) + ziel_pfad = rename_tool.find_dir_ci(store_pfad, ziel) if store_pfad else None + ziel_mails = rename_tool.count_mails(ziel_pfad) if ziel_pfad else 0 + + argumente = ["--adopt", "--domain", domain, + "--old-login", quelle, "--new-login", ziel] + + zeilen = [ + ("", f"Domain : {domain}"), + ("", f"Verzeichnis : {quelle}"), + ("", f"Neuer Besitzer: {ziel}"), + ("", ""), + ("hinweis", "users.cfg wird nicht geaendert - es wandert nur das Verzeichnis."), + ] + if ziel_mails: + argumente.append("--force") + zeilen += [ + ("", ""), + ("warnung", f"'{ziel}' hat bereits {ziel_mails} Mail(s) im Postfach."), + ("warnung", "Das bestehende Postfach wird umbenannt (.abgeloest-...)"), + ("warnung", "und bleibt vollstaendig erhalten - geloescht wird nichts."), + ] + zeilen += [("", ""), ("", "Gleichwertiger Aufruf:"), + ("hinweis", _befehl_text(argumente))] + + return argumente if bestaetige(win, styles, "Zusammenfassung", zeilen) else None + + +def ablauf_anzeigen(win, styles, config: UsersConfig, store_dir: Path) -> None: + zeilen: list[tuple[str, str]] = [] + for domain in config.domains(): + zeilen.append(("titel", f"Domain {domain}")) + for b in config.users(domain): + marken = [] + if not b["enabled"]: + marken.append("deaktiviert") + if not config.is_internal(b["loginName"], domain): + marken.append("Verzeichnisdienst") + zusatz = f" [{', '.join(marken)}]" if marken else "" + zeilen.append(("", f" {b['loginName']:<26} {b['fullName']}{zusatz}")) + + bekannt = {b["loginName"].lower() for b in config.users(domain)} + for name in verwaiste_verzeichnisse(store_dir, domain, bekannt): + zeilen.append(("warnung", + f" {name:<26} (verwaistes Verzeichnis, kein Benutzer)")) + zeilen.append(("", "")) + zeige_text(win, styles, "Benutzer laut users.cfg", zeilen) + + +# ========================================================================== +# Rahmenprogramm +# ========================================================================== + + +def oberflaeche(stdscr, config_pfad: Path, store_dir: Path) -> list[str] | None: + curses.curs_set(0) + styles = _setup_colors() + # keypad loest Pfeiltasten in KEY_UP/KEY_DOWN auf, statt die rohe + # Escape-Sequenz zu liefern - sonst laese die Oberflaeche das fuehrende + # ESC als "zurueck" und spraenge bei jedem Tastendruck heraus. + stdscr.keypad(True) + # Wartezeit, in der curses ein einzelnes ESC von einer Sequenz + # unterscheidet. Der Standard von einer Sekunde macht die Bedienung zaeh. + try: + curses.set_escdelay(25) + except (AttributeError, curses.error): + pass # aeltere Python-Versionen kennen die Funktion nicht + + punkte = [ + "Benutzer umbenennen (Login-Name und Mailbox-Verzeichnis)", + "Verwaistes Mailbox-Verzeichnis einem Benutzer zuordnen", + "Benutzer und verwaiste Verzeichnisse anzeigen", + "Beenden", + ] + + while True: + try: + config = UsersConfig(config_pfad) + except AbortedError as exc: + zeige_text(stdscr, styles, "Fehler", [("warnung", str(exc))]) + return None + + try: + wahl = menue(stdscr, styles, "Was moechten Sie tun?", punkte, + "Pfeiltasten Enter=Auswahl q=beenden") + except Abbruch: + return None + + try: + if wahl == 0: + argumente = ablauf_umbenennen(stdscr, styles, config, store_dir) + if argumente: + return argumente + elif wahl == 1: + argumente = ablauf_uebernehmen(stdscr, styles, config, store_dir) + if argumente: + return argumente + elif wahl == 2: + ablauf_anzeigen(stdscr, styles, config, store_dir) + else: + return None + except Abbruch: + continue # zurueck ins Hauptmenue + + +def main(argv: list[str]) -> int: + install_dir = Path(os.environ.get("KERIO_INSTALL_DIR", DEFAULT_INSTALL_DIR)) + store_dir = Path(os.environ.get("KERIO_STORE_DIR", DEFAULT_STORE_DIR)) + config_pfad = install_dir / "users.cfg" + + if not config_pfad.is_file(): + print(f"FEHLER: '{config_pfad}' nicht gefunden.\n" + "Laeuft dieses Programm auf dem Kerio-Server? Anderer Pfad " + "ueber die Umgebungsvariable KERIO_INSTALL_DIR.", file=sys.stderr) + return 2 + + if not sys.stdout.isatty(): + print("FEHLER: Der Assistent braucht ein Terminal.", file=sys.stderr) + return 2 + + try: + argumente = curses.wrapper(oberflaeche, config_pfad, store_dir) + except curses.error as exc: + print(f"FEHLER in der Anzeige: {exc}\n" + "Vielleicht ist das Fenster zu klein.", file=sys.stderr) + return 2 + + if not argumente: + print("Nichts geaendert.") + return 0 + + # Ab hier laeuft alles wieder im normalen Terminal, damit der Ablauf + # dieselbe Ausgabe liefert wie beim Aufruf ueber die Kommandozeile. + argumente = argumente + ["--install-dir", str(install_dir), + "--store-dir", str(store_dir), "--yes"] + print("Aufruf:", _befehl_text([a for a in argumente if a != "--yes"])) + print() + + if os.geteuid() != 0: + print("FEHLER: Dafuer werden root-Rechte benoetigt. Bitte mit sudo " + "starten.", file=sys.stderr) + return 2 + + return rename_tool.main(argumente) + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/tests/test_tui.py b/tests/test_tui.py new file mode 100644 index 0000000..1921fdf --- /dev/null +++ b/tests/test_tui.py @@ -0,0 +1,65 @@ +"""Testet die Logik des Assistenten, soweit sie ohne Terminal pruefbar ist.""" +import sys, tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from kerio_tui import verwaiste_verzeichnisse, _befehl_text + +ok = True + + +def check(label, cond): + global ok + print(f" {'ok ' if cond else 'FEHLER'} {label}") + if not cond: + ok = False + + +store = Path(tempfile.mkdtemp()) / "store" +for name in ("anna.schmidt", "bob", "alt.name", "ex.mitarbeiter", + "#public", "anna.schmidt.abgeloest-20260807-120000"): + (store / "mail" / "firma.de" / name).mkdir(parents=True) +# Mails nur im verwaisten Verzeichnis +msgs = store / "mail" / "firma.de" / "alt.name" / "INBOX" / "#msgs" +msgs.mkdir(parents=True) +for i in range(4): + (msgs / f"{i}.eml").write_text("x") +# ein Archiv-Verzeichnis ohne Gegenstueck in mail/ +(store / "archive" / "firma.de" / "nur.archiv").mkdir(parents=True) +# eine zweite Domain, die nicht hineinfunken darf +(store / "mail" / "andere.de" / "fremd.user").mkdir(parents=True) + +print("== Verwaiste Verzeichnisse erkennen ==") +bekannt = {"anna.schmidt", "bob"} +gefunden = verwaiste_verzeichnisse(store, "firma.de", bekannt) + +check("findet 'alt.name'", "alt.name" in gefunden) +check("findet 'ex.mitarbeiter'", "ex.mitarbeiter" in gefunden) +check("findet auch reine Archiv-Verzeichnisse", "nur.archiv" in gefunden) +check("uebergeht bestehende Benutzer", + "anna.schmidt" not in gefunden and "bob" not in gefunden) +check("uebergeht Kerio-Ordner (#public)", "#public" not in gefunden) +check("uebergeht bereits abgeloeste Sicherungen", + not any(".abgeloest-" in g for g in gefunden)) +check("greift nicht in andere Domains", "fremd.user" not in gefunden) +check("keine Doppelten", len(gefunden) == len(set(gefunden))) + +print("\n== Gross-/Kleinschreibung ==") +gefunden2 = verwaiste_verzeichnisse(store, "FIRMA.DE", {"ANNA.SCHMIDT", "bob"}) +check("Domain unabhaengig von Schreibweise", "alt.name" in gefunden2) +check("bekannte Logins unabhaengig von Schreibweise", + "anna.schmidt" not in gefunden2) + +print("\n== Domain ohne Store-Verzeichnis ==") +check("liefert leere Liste statt Fehler", + verwaiste_verzeichnisse(store, "gibtsnicht.de", set()) == []) + +print("\n== Befehlszeile fuer die Anzeige ==") +text = _befehl_text(["--domain", "firma.de", "--old-login", "alt.name", + "--new-login", "neu.name", "--full-name", "Neue Person"]) +check("beginnt mit dem Scriptnamen", text.startswith("sudo ./kerio_rename_user.py")) +check("Namen mit Leerzeichen werden gequotet", '"Neue Person"' in text) +check("Namen ohne Leerzeichen bleiben blank", " firma.de " in text) + +print("\n" + ("ALLE TESTS BESTANDEN" if ok else "FEHLGESCHLAGEN")) +sys.exit(0 if ok else 1)