Der Test gegen einen echten Server (Kerio Connect 10.0.9 patch 2) hat gezeigt, dass Users.set das Feld loginName stillschweigend ignoriert - kein Fehler, keine Wirkung, waehrend fullName im selben Aufruf sauber uebernommen wird. Vier Varianten geprueft (nur loginName, mit domainId, mit leeren emailAddresses, vollstaendiges User-Objekt): alle wirkungslos. Der bisherige Ansatz konnte also gar nicht funktionieren. Der Login-Name steht in users.cfg, und zwar in zwei Listen: User und UserAdditionalData. Beide verweisen ueber Name+Domain statt ueber die Guid, beide muessen mit, sonst verliert der Benutzer seine Passworthistorie. Geaendert wird der Rohtext, damit Formatierung und unbekannte Felder unangetastet bleiben. users.cfg und Store-Verzeichnis wandern jetzt in EINEM Stopp-Fenster. Laeuft Kerio zwischendurch mit nur einer Haelfte, legt es die vermeintlich fehlende Mailbox sofort neu an - beim Entwickeln genau so passiert. Damit entfaellt auch der Admin-Zugang: das Script braucht keine Zugangsdaten mehr, --list zeigt die Benutzer aus der Datei. Ausserdem gefunden: itemSource heisst 'DSInternalSource', nicht 'Internal' wie angenommen - die alte Pruefung haette bei jedem lokalen Benutzer faelschlich abgebrochen. Die Erkennung laeuft jetzt ueber InternalDb in users.cfg. Tests fuer beide Haelften unter tests/. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
428 lines
16 KiB
Python
Executable File
428 lines
16 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Benennt einen Benutzer in Kerio Connect vollstaendig um:
|
|
|
|
1. den Login-Namen in users.cfg (und optional den angezeigten Namen)
|
|
2. das zugehoerige Mailbox-Verzeichnis im Store auf der Platte
|
|
|
|
Warum nicht ueber die Administration API: Kerio nimmt loginName in Users.set
|
|
zwar entgegen, ignoriert das Feld aber stillschweigend - kein Fehler, keine
|
|
Wirkung (geprueft gegen Kerio Connect 10.0.9). Der Login-Name laesst sich nur
|
|
in users.cfg aendern. Deshalb kommt dieses Script ohne Admin-Zugangsdaten aus.
|
|
|
|
Kerio legt die Mailbox unter <store>/mail/<domain>/<loginname>/ ab. Ohne
|
|
Schritt 2 findet der Benutzer nach der Umbenennung eine leere Mailbox vor,
|
|
waehrend die alten Mails unter dem alten Verzeichnisnamen liegenbleiben.
|
|
|
|
Beide Aenderungen passieren in EINEM Stopp-Fenster des Dienstes. Das ist keine
|
|
Bequemlichkeit: laeuft Kerio zwischendurch mit nur einer der beiden Haelften,
|
|
legt es die fehlende Mailbox sofort neu an und man hat ein verwaistes
|
|
Verzeichnis mehr.
|
|
|
|
Das Script muss LOKAL AUF DEM KERIO-SERVER ALS ROOT laufen.
|
|
|
|
Beispiele:
|
|
|
|
./kerio_rename_user.py --list # Benutzer anzeigen
|
|
sudo ./kerio_rename_user.py \\
|
|
--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 os
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from kerio_common import (
|
|
DEFAULT_INSTALL_DIR,
|
|
DEFAULT_SERVICE,
|
|
DEFAULT_STORE_DIR,
|
|
AbortedError,
|
|
ServiceController,
|
|
)
|
|
from kerio_users_config import UsersConfig, validate_login
|
|
|
|
# Store-Unterbaeume, in denen pro Benutzer ein Verzeichnis liegen kann.
|
|
# Alle folgen dem Schema <store>/<subtree>/<domain>/<loginname>/
|
|
STORE_SUBTREES = ("mail", "archive")
|
|
|
|
BACKUP_SUFFIX = ".bak-rename-user"
|
|
|
|
|
|
# ==========================================================================
|
|
# 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))
|
|
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]]) -> list[tuple[Path, Path]]:
|
|
"""Fuehrt die Verschiebungen aus. Gibt zurueck, was tatsaechlich getan wurde."""
|
|
done: list[tuple[Path, Path]] = []
|
|
for source, target in renames:
|
|
try:
|
|
os.rename(source, target)
|
|
except OSError as exc:
|
|
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 list_users(config: UsersConfig, domain: str | None) -> int:
|
|
domains = config.domains()
|
|
if not domains:
|
|
print("Keine Benutzer in users.cfg gefunden.")
|
|
return 1
|
|
for name in domains:
|
|
if domain and name.lower() != domain.lower():
|
|
continue
|
|
print(f"\nDomain: {name}")
|
|
for user in config.users(name):
|
|
flags = [] if user["enabled"] else ["deaktiviert"]
|
|
if not config.is_internal(user["loginName"], name):
|
|
flags.append("Verzeichnisdienst")
|
|
suffix = f" [{', '.join(flags)}]" if flags else ""
|
|
print(f" {user['loginName']:<28} {user['fullName']}{suffix}")
|
|
return 0
|
|
|
|
|
|
def rename_user(args: argparse.Namespace, config: UsersConfig,
|
|
service: ServiceController) -> int:
|
|
store_dir = Path(args.store_dir)
|
|
|
|
# -- Vorpruefungen: nichts wird geschrieben ---------------------------
|
|
domains = config.domains()
|
|
domain = next((d for d in domains if d.lower() == args.domain.lower()), None)
|
|
if domain is None:
|
|
raise AbortedError(
|
|
f"Domain '{args.domain}' kommt in users.cfg nicht vor. "
|
|
f"Vorhanden: {', '.join(domains) or '(keine)'}"
|
|
)
|
|
|
|
user = config.find_user(args.old_login, domain)
|
|
if user is None:
|
|
raise AbortedError(
|
|
f"Benutzer '{args.old_login}' existiert nicht in Domain '{domain}'."
|
|
)
|
|
|
|
old_login = user["loginName"]
|
|
new_login = args.new_login or old_login
|
|
if new_login != old_login:
|
|
validate_login(new_login)
|
|
|
|
if not config.is_internal(old_login, domain) and not args.force:
|
|
raise AbortedError(
|
|
"Benutzer stammt aus einem Verzeichnisdienst (LDAP/AD). Der "
|
|
"Login-Name muss dort geaendert werden, nicht in Kerio. "
|
|
"Mit --force trotzdem versuchen."
|
|
)
|
|
|
|
if new_login.lower() != old_login.lower() and \
|
|
config.find_user(new_login, domain) is not None:
|
|
raise AbortedError(f"Login-Name '{new_login}' ist bereits vergeben.")
|
|
|
|
renames: list[tuple[Path, Path]] = []
|
|
if new_login != old_login:
|
|
renames = plan_store_renames(store_dir, domain, old_login, new_login)
|
|
if not renames:
|
|
print(f" Hinweis: kein Mailbox-Verzeichnis fuer '{old_login}' gefunden. "
|
|
"Es wird nur users.cfg geaendert.")
|
|
check_rename_targets(renames)
|
|
service.preflight()
|
|
|
|
needs_name = new_login != old_login
|
|
needs_fullname = bool(args.full_name) and args.full_name != user["fullName"]
|
|
if not needs_name and not needs_fullname:
|
|
print("\nNichts zu tun.")
|
|
return 0
|
|
|
|
# -- Plan ausgeben ----------------------------------------------------
|
|
print("\nGeplante Aenderungen")
|
|
print("--------------------")
|
|
print(f" Domain : {domain}")
|
|
print(f" Guid : {user['guid']}")
|
|
|
|
if needs_name:
|
|
print(f" Login-Name : {old_login} -> {new_login}")
|
|
print(f" Adresse : {old_login}@{domain} -> {new_login}@{domain}")
|
|
else:
|
|
print(f" Login-Name : {old_login} (unveraendert)")
|
|
|
|
if needs_fullname:
|
|
print(f" Voller Name : {user['fullName']} -> {args.full_name}")
|
|
else:
|
|
print(f" Voller Name : {user['fullName']} (unveraendert)")
|
|
|
|
print(f" users.cfg : {config.path}")
|
|
if renames:
|
|
print(f" Dienst : {service.describe()}")
|
|
for source, target in renames:
|
|
print(f" Verzeichnis : {source}")
|
|
print(f" -> {target}")
|
|
|
|
extra = config.other_occurrences(old_login, domain) if needs_name else []
|
|
if extra:
|
|
print("\n ACHTUNG: Der alte Name kommt auch hier vor und wird NICHT "
|
|
"automatisch geaendert:")
|
|
for hit in extra:
|
|
print(f" - {hit}")
|
|
print(" Bitte nach der Umbenennung in der Administrationskonsole pruefen.")
|
|
|
|
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.")
|
|
if input("Aenderungen jetzt durchfuehren? [j/N] ").strip().lower() \
|
|
not in ("j", "ja", "y", "yes"):
|
|
print("Abgebrochen.")
|
|
return 1
|
|
|
|
# -- 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}")
|
|
|
|
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)
|
|
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()
|
|
|
|
# -- Gegenlesen -------------------------------------------------------
|
|
check = UsersConfig(config.path)
|
|
result = check.find_user(new_login, domain)
|
|
if result is None:
|
|
print(f"\n WARNUNG: '{new_login}' ist in users.cfg nach dem Start nicht "
|
|
"auffindbar. Bitte pruefen.", file=sys.stderr)
|
|
return 2
|
|
|
|
print(f"\nFertig. Benutzer heisst jetzt '{result['loginName']}' "
|
|
f"({result['fullName']}).")
|
|
if done:
|
|
print(f"Mailbox liegt unter '{done[0][1]}'.")
|
|
print("Hinweis: Der Benutzer muss sich in allen Mail-Clients mit dem "
|
|
"neuen Login neu anmelden.")
|
|
if needs_name:
|
|
print("Die alte Adresse nimmt keine Mail mehr an. Wer das braucht, legt "
|
|
"in der Administrationskonsole einen Alias an.")
|
|
return 0
|
|
|
|
|
|
# ==========================================================================
|
|
# 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("--install-dir", default=DEFAULT_INSTALL_DIR,
|
|
help=f"Kerio-Installationsverzeichnis "
|
|
f"(Standard: {DEFAULT_INSTALL_DIR})")
|
|
parser.add_argument("--store-dir", default=DEFAULT_STORE_DIR,
|
|
help=f"Kerio-Store-Verzeichnis (Standard: {DEFAULT_STORE_DIR})")
|
|
parser.add_argument("--domain",
|
|
help="Mail-Domain des Benutzers, z. B. firma.de")
|
|
parser.add_argument("--old-login",
|
|
help="Bisheriger Login-Name (ohne @domain)")
|
|
parser.add_argument("--new-login",
|
|
help="Neuer Login-Name (ohne @domain)")
|
|
parser.add_argument("--full-name",
|
|
help='Neuer angezeigter Name, z. B. "Anna Schmidt"')
|
|
parser.add_argument("--list", action="store_true",
|
|
help="Benutzer aus users.cfg auflisten und beenden")
|
|
|
|
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 sein.")
|
|
parser.add_argument("--service-wait", type=int, default=90,
|
|
help="Sekunden Wartezeit aufs Stoppen (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 Benutzern aus einem Verzeichnisdienst")
|
|
|
|
args = parser.parse_args(argv)
|
|
|
|
if not args.list:
|
|
missing = [n for n in ("domain", "old_login") if not getattr(args, n)]
|
|
if missing:
|
|
parser.error("Benoetigt: " + ", ".join("--" + m.replace("_", "-")
|
|
for m in missing))
|
|
if not args.new_login and not args.full_name:
|
|
parser.error("Mindestens --new-login oder --full-name angeben.")
|
|
if not args.dry_run and os.geteuid() != 0:
|
|
parser.error(
|
|
"Zum Aendern von users.cfg und zum Verschieben des "
|
|
"Store-Verzeichnisses werden root-Rechte benoetigt. "
|
|
"Mit sudo starten (oder --dry-run zum Testen)."
|
|
)
|
|
return args
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
args = parse_args(argv)
|
|
config_path = Path(args.install_dir) / "users.cfg"
|
|
|
|
service = ServiceController(
|
|
name=args.service_name,
|
|
mode=args.service_manager,
|
|
wait_seconds=args.service_wait,
|
|
dry_run=args.dry_run,
|
|
)
|
|
|
|
try:
|
|
config = UsersConfig(config_path)
|
|
if args.list:
|
|
return list_users(config, args.domain)
|
|
return rename_user(args, config, service)
|
|
except AbortedError as exc:
|
|
print(f"\nFEHLER: {exc}", file=sys.stderr)
|
|
return 2
|
|
except PermissionError as exc:
|
|
print(f"\nFEHLER: Keine Berechtigung: {exc}\nMit sudo starten.",
|
|
file=sys.stderr)
|
|
return 2
|
|
except KeyboardInterrupt:
|
|
print("\nAbgebrochen.", file=sys.stderr)
|
|
return 130
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|