Hacker-Software/kerio-connect-rename-user
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
kerio-connect-rename-user/kerio_tui.py
T
duffyduckandClaude Opus 5 24070cba4e Assistent mit Menuefuehrung, dazu zwei Haertungen 
kerio_tui.py fuehrt durch Umbenennung und Uebernahme, ohne dass man
Parameter kennen muss: Benutzer und Domains kommen zur Auswahl, verwaiste
Mailbox-Verzeichnisse findet der Assistent selbst, und vor dem Ausfuehren
zeigt er den gleichwertigen Kommandozeilenaufruf. Nur curses aus der
Standardbibliothek, keine neue Abhaengigkeit.

Beim Durchfahren ueber ein PTY kamen zwei Dinge heraus:

keypad(True) loest die Pfeiltasten auf diesem Server nicht auf - getch()
liefert 27, 91, 66 statt KEY_DOWN, wofuer ein unpassendes terminfo
genuegt. Das fuehrende ESC haette als "zurueck" gegolten und den
Assistenten bei jedem Pfeildruck verlassen. Er loest die Sequenzen jetzt
selbst auf; ein einzelnes ESC wirkt weiterhin als Abbruch.

Und: Bricht die Verbindung weg, waehrend Kerio gestoppt ist, stirbt das
Script an SIGHUP und laesst den Dienst unten - genau so passiert, als der
Testtreiber die Verbindung schloss. ServiceController.abgesichert() faengt
SIGHUP, SIGTERM und SIGINT ab und startet den Dienst wieder, bevor der
Abbruch durchgereicht wird. Gegen den Testserver geprueft: Signal mitten
im Stopp-Fenster, Dienst kommt hoch, Daten unberuehrt.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-08-07 14:54:33 +02:00

546 lines
20 KiB
Python
Raw Permalink Blame History

#!/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:]))
Reference in New Issue View Git Blame Copy Permalink