Hacker-Software/kerio-connect-rename-user
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b12041fc8110bffd590eb5cbe76a626226db775d
kerio-connect-rename-user/kerio_rename_user.py
T
duffyduckandClaude Opus 5 b12041fc81 Script fuer den Erstkonfigurationsassistenten, gemeinsames Modul 
kerio_trial_setup.py bringt eine frische Installation am Assistenten
vorbei, wenn dessen Lizenzschritt nicht durchlaufbar ist: der Link
"Im unregistrierten Modus aktivieren" fehlt je nach Version und die
verlinkte Registrierungsseite ist seit der GFI-Uebernahme teils tot.
Gesetzt wird ConfigWizardDone in mailserver.cfg; Kerio laeuft danach
30 Tage unregistriert, ohne Einschraenkung bei Benutzerverwaltung und
Admin-API.

Die Datei darf nur bei gestopptem Dienst bearbeitet werden, weil Kerio
seine Konfiguration beim Beenden aus dem Speicher zurueckschreibt. Das
Script stoppt den Dienst, liest danach neu ein, schreibt atomar ueber
eine temporaere Datei und liest den Wert nach dem Start zur Kontrolle
erneut. Backup vor jeder Aenderung, --restore spielt es zurueck.

API-Client und Dienststeuerung wandern nach kerio_common.py, damit
beide Scripts sie teilen.

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

408 lines
15 KiB
Python
Executable File
Raw Blame History

#!/usr/bin/env python3
"""
Benennt einen Benutzer in Kerio Connect vollstaendig um:
1. Login-Name (und optional voller Name) ueber die Administration API
2. das zugehoerige Mailbox-Verzeichnis im Store auf der Platte
Kerio legt die Mailbox unter <store>/mail/<domain>/<loginname>/ ab und zieht
das Verzeichnis beim Aendern des Login-Namens nicht mit. Ohne Schritt 2 findet
der Benutzer nach der Umbenennung eine leere Mailbox vor.
Damit das Verzeichnis gefahrlos verschoben werden kann, wird der Kerio-Dienst
waehrend des Umbenennens gestoppt und danach wieder gestartet.
Das Script muss deshalb LOKAL AUF DEM KERIO-SERVER ALS ROOT laufen.
Beispiel:
sudo ./kerio_rename_user.py \\
--admin admin \\
--domain firma.de \\
--old-login anna.mueller \\
--new-login anna.schmidt \\
--full-name "Anna Schmidt" \\
--dry-run
Ohne --dry-run wird vor dem Schreiben nachgefragt (ausser mit --yes).
"""
from __future__ import annotations
import argparse
import getpass
import json
import os
import sys
import time
from pathlib import Path
from kerio_common import (
DEFAULT_SERVICE,
DEFAULT_STORE_DIR,
AbortedError,
KerioAdminApi,
KerioError,
ServiceController,
find_domain,
find_user,
)
# Store-Unterbaeume, in denen pro Benutzer ein Verzeichnis liegen kann.
# Alle folgen dem Schema <store>/<subtree>/<domain>/<loginname>/
STORE_SUBTREES = ("mail", "archive")
# ==========================================================================
# Store-Verzeichnisse
# ==========================================================================
def find_dir_ci(parent: Path, name: str) -> Path | None:
"""Sucht ein Unterverzeichnis - erst exakt, dann ohne Ruecksicht auf Gross/Klein.
Kerio legt die Verzeichnisse meist klein geschrieben an, auch wenn der
Login-Name gemischt geschrieben ist.
"""
direct = parent / name
if direct.is_dir():
return direct
if not parent.is_dir():
return None
wanted = name.lower()
for child in parent.iterdir():
if child.is_dir() and child.name.lower() == wanted:
return child
return None
def plan_store_renames(store_dir: Path, domain_name: str, old_login: str,
new_login: str) -> list[tuple[Path, Path]]:
"""Ermittelt alle zu verschiebenden Verzeichnisse als (quelle, ziel)-Paare."""
if not store_dir.is_dir():
raise AbortedError(
f"Store-Verzeichnis '{store_dir}' existiert nicht. "
"Richtigen Pfad mit --store-dir angeben."
)
renames: list[tuple[Path, Path]] = []
for subtree in STORE_SUBTREES:
domain_dir = find_dir_ci(store_dir / subtree, domain_name)
if domain_dir is None:
continue
source = find_dir_ci(domain_dir, old_login)
if source is None:
continue
# Schreibweise des vorhandenen Verzeichnisses uebernehmen.
target_name = new_login.lower() if source.name.islower() else new_login
renames.append((source, domain_dir / target_name))
if not renames:
raise AbortedError(
f"Kein Mailbox-Verzeichnis fuer '{old_login}' unter {store_dir} gefunden "
f"(gesucht in: {', '.join(STORE_SUBTREES)}/{domain_name}/). "
"Pfad mit --store-dir pruefen."
)
return renames
def check_rename_targets(renames: list[tuple[Path, Path]]) -> None:
"""Stellt sicher, dass kein Ziel bereits belegt ist."""
for source, target in renames:
if target.exists() and target.resolve() != source.resolve():
raise AbortedError(
f"Zielverzeichnis '{target}' existiert bereits. "
"Bitte manuell pruefen und wegraeumen."
)
try:
same_fs = source.stat().st_dev == target.parent.stat().st_dev
except OSError as exc:
raise AbortedError(f"'{source}' ist nicht mehr lesbar: {exc}") from exc
if not same_fs:
raise AbortedError(
f"'{source}' und '{target.parent}' liegen auf verschiedenen "
"Dateisystemen - das wuerde ein Kopieren statt Verschieben bedeuten."
)
def apply_store_renames(renames: list[tuple[Path, Path]],
dry_run: bool) -> list[tuple[Path, Path]]:
"""Fuehrt die Verschiebungen aus. Gibt zurueck, was tatsaechlich getan wurde."""
done: list[tuple[Path, Path]] = []
for source, target in renames:
if dry_run:
print(f" [dry-run] wuerde verschieben: {source} -> {target}")
continue
try:
os.rename(source, target)
except OSError as exc:
# Bereits erfolgte Verschiebungen zuruecknehmen.
_revert_store_renames(done)
raise AbortedError(
f"Verschieben von '{source}' nach '{target}' fehlgeschlagen: {exc}"
) from exc
done.append((source, target))
print(f" Verschoben: {source} -> {target}")
return done
def _revert_store_renames(done: list[tuple[Path, Path]]) -> None:
for source, target in reversed(done):
try:
os.rename(target, source)
print(f" Zurueckgenommen: {target} -> {source}", file=sys.stderr)
except OSError as exc:
print(f" WARNUNG: Rueckgaengigmachen von '{target}' fehlgeschlagen: {exc}",
file=sys.stderr)
# ==========================================================================
# Ablauf
# ==========================================================================
def rename_user(api: KerioAdminApi, args: argparse.Namespace,
service: ServiceController) -> int:
store_dir = Path(args.store_dir)
# -- Vorpruefungen: nichts wird geschrieben ---------------------------
domain = find_domain(api, args.domain)
domain_id = domain["id"]
domain_name = domain["name"]
print(f"Domain: {domain_name} (id={domain_id})")
user = find_user(api, domain_id, args.old_login)
if user is None:
raise KerioError(
f"Benutzer '{args.old_login}' existiert nicht in Domain '{domain_name}'."
)
old_login = user["loginName"]
new_login = args.new_login or old_login
item_source = user.get("itemSource")
if item_source and item_source.lower() != "internal":
message = (
f"Benutzer stammt aus '{item_source}' (Verzeichnisdienst/LDAP). "
"Der Login-Name muss dort geaendert werden, nicht in Kerio."
)
if not args.force:
raise AbortedError(message + " Mit --force trotzdem versuchen.")
print(f" WARNUNG: {message}")
if new_login != old_login and find_user(api, domain_id, new_login) is not None:
raise AbortedError(f"Login-Name '{new_login}' ist bereits vergeben.")
pattern: dict = {}
if new_login != old_login:
pattern["loginName"] = new_login
if args.full_name and args.full_name != user.get("fullName"):
pattern["fullName"] = args.full_name
renames: list[tuple[Path, Path]] = []
if new_login != old_login:
renames = plan_store_renames(store_dir, domain_name, old_login, new_login)
check_rename_targets(renames)
service.preflight()
if not pattern and not renames:
print("\nNichts zu tun.")
return 0
# -- Plan ausgeben ----------------------------------------------------
print("\nGeplante Aenderungen")
print("--------------------")
print(f" Benutzer-ID : {user['id']}")
if "loginName" in pattern:
print(f" Login-Name : {old_login} -> {new_login}")
else:
print(f" Login-Name : {old_login} (unveraendert)")
old_full_name = user.get("fullName", "")
if "fullName" in pattern:
print(f" Voller Name : {old_full_name} -> {args.full_name}")
else:
print(f" Voller Name : {old_full_name} (unveraendert)")
print(f" Adresse : {old_login}@{domain_name} -> {new_login}@{domain_name}")
if renames:
print(f" Dienst : {service.describe()}")
for source, target in renames:
print(f" Verzeichnis : {source}")
print(f" -> {target}")
addresses = user.get("emailAddresses") or []
if addresses:
print(f" Weitere Adressen am Benutzer (unveraendert): "
f"{json.dumps(addresses, ensure_ascii=False)}")
if args.dry_run:
print("\n--dry-run: es wurde nichts geaendert.")
return 0
if not args.yes:
print("\nAchtung: der Kerio-Dienst wird dabei kurz gestoppt.")
answer = input("Aenderungen jetzt durchfuehren? [j/N] ").strip().lower()
if answer not in ("j", "ja", "y", "yes"):
print("Abgebrochen.")
return 1
# -- Schreiben --------------------------------------------------------
print()
if pattern:
api.call("Users.set", {"userIds": [user["id"]], "pattern": pattern})
print(f" Benutzer aktualisiert: {json.dumps(pattern, ensure_ascii=False)}")
if not renames:
print("\nFertig (keine Verzeichnisse zu verschieben).")
return 0
# Ab hier ist die API-Aenderung schon durch. Faellt das Verschieben aus,
# wird sie zurueckgedreht, damit Konfiguration und Platte zusammenpassen.
api.logout()
try:
service.stop()
# Nach dem Stoppen erneut pruefen: Kerio koennte in der Zwischenzeit
# ein leeres Verzeichnis unter dem neuen Namen angelegt haben.
check_rename_targets(renames)
apply_store_renames(renames, dry_run=False)
except AbortedError as exc:
print(f"\nFEHLER beim Verschieben: {exc}", file=sys.stderr)
_rollback_api_change(args, user, old_login, service)
return 2
finally:
service.start()
print(f"\nFertig. Benutzer heisst jetzt '{new_login}', "
f"Mailbox liegt unter '{renames[0][1]}'.")
print("Hinweis: Der Benutzer muss sich in allen Mail-Clients mit dem "
"neuen Login neu anmelden.")
return 0
def _rollback_api_change(args: argparse.Namespace, user: dict, old_login: str,
service: ServiceController) -> None:
"""Setzt den Login-Namen zurueck, wenn das Verschieben gescheitert ist."""
print(" Setze Login-Namen zurueck ...", file=sys.stderr)
service.start()
# Nach dem Start braucht die Admin-API einen Moment, bis sie antwortet.
last_error: Exception | None = None
for attempt in range(10):
try:
with KerioAdminApi(args.server, args.port,
verify_tls=not args.insecure) as api:
api.login(args.admin, args.password)
api.call("Users.set", {
"userIds": [user["id"]],
"pattern": {"loginName": old_login},
})
print(f" Login-Name wieder auf '{old_login}' gesetzt.", file=sys.stderr)
return
except KerioError as exc:
last_error = exc
time.sleep(3)
print(f" WARNUNG: Zuruecksetzen fehlgeschlagen: {last_error}\n"
f" Der Benutzer heisst in Kerio jetzt '{args.new_login}', "
f"die Mailbox liegt aber noch unter '{old_login}'. "
"Bitte von Hand richten!", file=sys.stderr)
# ==========================================================================
# CLI
# ==========================================================================
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Benennt einen Kerio-Connect-Benutzer inklusive Store-Verzeichnis um.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__,
)
parser.add_argument("--server", default="localhost",
help="Hostname der Admin-API (Standard: localhost)")
parser.add_argument("--port", type=int, default=4040,
help="Port der Admin-API (Standard: 4040)")
parser.add_argument("--admin", required=True,
help="Benutzername des Administrators")
parser.add_argument("--password",
help="Admin-Passwort. Besser: Umgebungsvariable "
"KERIO_ADMIN_PASSWORD oder interaktive Eingabe.")
parser.add_argument("--domain", required=True,
help="Mail-Domain des Benutzers, z. B. firma.de")
parser.add_argument("--old-login", required=True,
help="Bisheriger Login-Name (ohne @domain)")
parser.add_argument("--new-login",
help="Neuer Login-Name (ohne @domain)")
parser.add_argument("--full-name",
help='Neuer voller Name, z. B. "Anna Schmidt"')
parser.add_argument("--store-dir", default=DEFAULT_STORE_DIR,
help=f"Kerio-Store-Verzeichnis (Standard: {DEFAULT_STORE_DIR})")
parser.add_argument("--service-name", default=DEFAULT_SERVICE,
help=f"Name des Kerio-Dienstes (Standard: {DEFAULT_SERVICE})")
parser.add_argument("--service-manager", default="auto",
choices=("auto", "systemd", "initd", "none"),
help="Wie der Dienst gestoppt wird. 'none' fasst ihn nicht an - "
"dann muss Kerio selbst gestoppt werden.")
parser.add_argument("--service-wait", type=int, default=90,
help="Sekunden, die auf das Stoppen gewartet wird (Standard: 90)")
parser.add_argument("--dry-run", action="store_true",
help="Nur anzeigen, was passieren wuerde")
parser.add_argument("--yes", action="store_true",
help="Rueckfrage ueberspringen")
parser.add_argument("--force", action="store_true",
help="Auch bei LDAP-/AD-gemappten Benutzern versuchen")
parser.add_argument("--insecure", action="store_true",
help="TLS-Zertifikat nicht pruefen (selbstsigniertes Zertifikat)")
args = parser.parse_args(argv)
if not args.new_login and not args.full_name:
parser.error("Mindestens --new-login oder --full-name angeben.")
if args.new_login and not args.dry_run and os.geteuid() != 0:
parser.error(
"Zum Verschieben des Store-Verzeichnisses werden root-Rechte benoetigt. "
"Mit sudo starten (oder --dry-run zum Testen)."
)
if not args.password:
args.password = os.environ.get("KERIO_ADMIN_PASSWORD")
if not args.password:
args.password = getpass.getpass(f"Passwort fuer {args.admin}: ")
return args
def main(argv: list[str]) -> int:
args = parse_args(argv)
service = ServiceController(
name=args.service_name,
mode=args.service_manager,
wait_seconds=args.service_wait,
dry_run=args.dry_run,
)
try:
with KerioAdminApi(args.server, args.port,
verify_tls=not args.insecure) as api:
api.login(args.admin, args.password)
print(f"Angemeldet an {args.server}:{args.port} als {args.admin}")
return rename_user(api, args, service)
except (KerioError, AbortedError) as exc:
print(f"\nFEHLER: {exc}", file=sys.stderr)
return 2
except KeyboardInterrupt:
print("\nAbgebrochen.", file=sys.stderr)
return 130
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))
Reference in New Issue View Git Blame Copy Permalink