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>
This commit is contained in:
duffyduck
co-authored by
Claude Opus 5
Claude Opus 5
parent
b9cbf65fac
commit
24070cba4e
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
+76
-68
@@ -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()
|
||||
|
||||
+545
@@ -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:]))
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user