duffyduckandClaude Opus 5
485a681a0a
Benennt einen Benutzer vollstaendig um: Login-Name und voller Name ueber die Administration API, dazu das Mailbox-Verzeichnis im Store auf der Platte. Kerio zieht das Verzeichnis beim Aendern des Login-Namens nicht mit, sodass der Benutzer sonst eine leere Mailbox vorfindet. Ablauf: Vorpruefungen (Benutzer vorhanden, neuer Name frei, kein LDAP-Benutzer, Verzeichnis vorhanden und Ziel frei, gleiches Dateisystem, Dienst steuerbar), dann Users.set, Dienst stoppen, Verzeichnisse verschieben, Dienst starten. Verschoben werden mail/ und archive/. Scheitert das Verschieben, werden bereits verschobene Verzeichnisse zurueckgenommen und der Login-Name zurueckgesetzt, damit Konfiguration und Platte konsistent bleiben. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
684 lines
25 KiB
Python
Executable File
684 lines
25 KiB
Python
Executable File
#!/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 http.cookiejar
|
|
import json
|
|
import os
|
|
import shutil
|
|
import ssl
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
|
|
APP_INFO = {
|
|
"name": "kerio-connect-user-rename",
|
|
"vendor": "inhouse",
|
|
"version": "2.0",
|
|
}
|
|
|
|
# Seitengroesse fuer die Paginierung von Users.get / Domains.get.
|
|
PAGE_SIZE = 500
|
|
|
|
DEFAULT_STORE_DIR = "/opt/kerio/mailserver/store"
|
|
DEFAULT_SERVICE = "kerio-connect"
|
|
|
|
# Store-Unterbaeume, in denen pro Benutzer ein Verzeichnis liegen kann.
|
|
# Alle folgen dem Schema <store>/<subtree>/<domain>/<loginname>/
|
|
STORE_SUBTREES = ("mail", "archive")
|
|
|
|
|
|
class KerioError(Exception):
|
|
"""Fehler, den der Kerio-Server gemeldet hat (oder ein Transportfehler)."""
|
|
|
|
|
|
class AbortedError(Exception):
|
|
"""Der Benutzer hat abgebrochen oder eine Vorpruefung ist fehlgeschlagen."""
|
|
|
|
|
|
# ==========================================================================
|
|
# Teil 1: Administration API (JSON-RPC)
|
|
# ==========================================================================
|
|
|
|
|
|
class KerioAdminApi:
|
|
"""Duenner JSON-RPC-Client fuer die Kerio Connect Admin-API.
|
|
|
|
Die API will nach dem Login drei Dinge sehen: das Session-Cookie, den
|
|
X-Token-Header und denselben Token nochmal im JSON-Payload.
|
|
"""
|
|
|
|
def __init__(self, host: str, port: int = 4040, verify_tls: bool = True,
|
|
timeout: int = 30) -> None:
|
|
self.url = f"https://{host}:{port}/admin/api/jsonrpc/"
|
|
self.timeout = timeout
|
|
self.token: str | None = None
|
|
self._request_id = 0
|
|
|
|
context = ssl.create_default_context()
|
|
if not verify_tls:
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
|
|
self._opener = urllib.request.build_opener(
|
|
urllib.request.HTTPSHandler(context=context),
|
|
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()),
|
|
)
|
|
|
|
# -- Transport ---------------------------------------------------------
|
|
|
|
def call(self, method: str, params: dict | None = None) -> dict:
|
|
self._request_id += 1
|
|
payload: dict = {"jsonrpc": "2.0", "id": self._request_id, "method": method}
|
|
if params is not None:
|
|
payload["params"] = params
|
|
if self.token:
|
|
payload["token"] = self.token
|
|
|
|
headers = {"Content-Type": "application/json-rpc; charset=UTF-8"}
|
|
if self.token:
|
|
headers["X-Token"] = self.token
|
|
|
|
request = urllib.request.Request(
|
|
self.url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
|
|
try:
|
|
with self._opener.open(request, timeout=self.timeout) as response:
|
|
body = json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as exc:
|
|
raise KerioError(f"HTTP {exc.code} bei {method}: {exc.read()[:500]!r}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise KerioError(f"Verbindung zu {self.url} fehlgeschlagen: {exc.reason}") from exc
|
|
except ssl.SSLError as exc:
|
|
raise KerioError(
|
|
f"TLS-Fehler: {exc}. Bei selbstsigniertem Zertifikat --insecure verwenden."
|
|
) from exc
|
|
|
|
if "error" in body:
|
|
error = body["error"]
|
|
raise KerioError(
|
|
f"{method}: {error.get('message', 'unbekannter Fehler')} "
|
|
f"(Code {error.get('code')})"
|
|
)
|
|
|
|
result = body.get("result", {})
|
|
_raise_on_error_list(method, result)
|
|
return result
|
|
|
|
def call_paged(self, method: str, params: dict, list_key: str = "list") -> list[dict]:
|
|
"""Ruft eine get-Methode so oft auf, bis alle totalItems eingesammelt sind."""
|
|
items: list[dict] = []
|
|
while True:
|
|
query = dict(params.get("query") or {})
|
|
query.update({"start": len(items), "limit": PAGE_SIZE})
|
|
query.setdefault("fields", [])
|
|
query.setdefault("conditions", [])
|
|
query.setdefault("combining", "Or")
|
|
|
|
result = self.call(method, {**params, "query": query})
|
|
page = result.get(list_key) or []
|
|
items.extend(page)
|
|
|
|
total = result.get("totalItems", len(items))
|
|
if not page or len(items) >= total:
|
|
return items
|
|
|
|
# -- Session -----------------------------------------------------------
|
|
|
|
def login(self, username: str, password: str) -> None:
|
|
result = self.call("Session.login", {
|
|
"userName": username,
|
|
"password": password,
|
|
"application": APP_INFO,
|
|
})
|
|
self.token = result.get("token")
|
|
if not self.token:
|
|
raise KerioError("Login lieferte keinen Token zurueck.")
|
|
|
|
def logout(self) -> None:
|
|
if self.token:
|
|
try:
|
|
self.call("Session.logout")
|
|
except KerioError:
|
|
pass # Ausloggen ist best effort
|
|
finally:
|
|
self.token = None
|
|
|
|
def __enter__(self) -> "KerioAdminApi":
|
|
return self
|
|
|
|
def __exit__(self, *_exc_info) -> None:
|
|
self.logout()
|
|
|
|
|
|
def _raise_on_error_list(method: str, result: dict) -> None:
|
|
"""Kerio liefert Teilfehler nicht als JSON-RPC-error, sondern als 'errors'-Liste."""
|
|
errors = result.get("errors")
|
|
if not errors:
|
|
return
|
|
details = "; ".join(
|
|
f"{e.get('message', '?')} (Code {e.get('code')}, Feld {e.get('inputIndex')})"
|
|
for e in errors
|
|
)
|
|
raise KerioError(f"{method} meldet Fehler: {details}")
|
|
|
|
|
|
def find_domain(api: KerioAdminApi, domain_name: str) -> dict:
|
|
domains = api.call_paged("Domains.get", {})
|
|
wanted = domain_name.strip().lower()
|
|
for domain in domains:
|
|
if domain.get("name", "").lower() == wanted:
|
|
return domain
|
|
known = ", ".join(sorted(d.get("name", "?") for d in domains)) or "(keine)"
|
|
raise KerioError(f"Domain '{domain_name}' nicht gefunden. Vorhanden: {known}")
|
|
|
|
|
|
def find_user(api: KerioAdminApi, domain_id: str, login_name: str) -> dict | None:
|
|
users = api.call_paged("Users.get", {"domainId": domain_id})
|
|
wanted = login_name.strip().lower()
|
|
for user in users:
|
|
if user.get("loginName", "").lower() == wanted:
|
|
return user
|
|
return None
|
|
|
|
|
|
# ==========================================================================
|
|
# Teil 2: Kerio-Dienst steuern
|
|
# ==========================================================================
|
|
|
|
|
|
class ServiceController:
|
|
"""Startet und stoppt den Kerio-Connect-Dienst.
|
|
|
|
Modus 'none' fasst den Dienst nicht an - dann muss man selbst dafuer
|
|
sorgen, dass Kerio waehrend des Verschiebens steht.
|
|
"""
|
|
|
|
def __init__(self, name: str, mode: str, wait_seconds: int = 90,
|
|
dry_run: bool = False) -> None:
|
|
self.name = name
|
|
self.wait_seconds = wait_seconds
|
|
self.dry_run = dry_run
|
|
self.mode = self._resolve_mode(mode)
|
|
|
|
@staticmethod
|
|
def _resolve_mode(mode: str) -> str:
|
|
if mode != "auto":
|
|
return mode
|
|
if shutil.which("systemctl"):
|
|
return "systemd"
|
|
if Path("/etc/init.d", DEFAULT_SERVICE).exists():
|
|
return "initd"
|
|
return "none"
|
|
|
|
def _run(self, *command: str) -> subprocess.CompletedProcess:
|
|
return subprocess.run(command, capture_output=True, text=True, timeout=120)
|
|
|
|
def describe(self) -> str:
|
|
if self.mode == "none":
|
|
return "Dienst wird nicht angefasst (--service-manager none)"
|
|
return f"{self.mode}: {self.name}"
|
|
|
|
def is_running(self) -> bool | None:
|
|
"""True/False, oder None wenn der Status nicht ermittelbar ist."""
|
|
if self.mode == "systemd":
|
|
return self._run("systemctl", "is-active", "--quiet", self.name).returncode == 0
|
|
if self.mode == "initd":
|
|
result = self._run(f"/etc/init.d/{self.name}", "status")
|
|
return result.returncode == 0
|
|
return None
|
|
|
|
def preflight(self) -> None:
|
|
"""Prueft vor allen Aenderungen, ob der Dienst ueberhaupt steuerbar ist."""
|
|
if self.mode == "none":
|
|
return
|
|
if self.mode == "systemd":
|
|
result = self._run("systemctl", "cat", self.name)
|
|
if result.returncode != 0:
|
|
raise AbortedError(
|
|
f"systemd kennt die Unit '{self.name}' nicht. "
|
|
"Richtigen Namen mit --service-name angeben oder "
|
|
"--service-manager none verwenden."
|
|
)
|
|
elif self.mode == "initd":
|
|
if not Path(f"/etc/init.d/{self.name}").exists():
|
|
raise AbortedError(f"/etc/init.d/{self.name} existiert nicht.")
|
|
|
|
def stop(self) -> None:
|
|
if self.mode == "none":
|
|
print(" Dienst: uebersprungen (--service-manager none)")
|
|
return
|
|
if self.dry_run:
|
|
print(f" [dry-run] wuerde Dienst '{self.name}' stoppen")
|
|
return
|
|
|
|
print(f" Stoppe Dienst '{self.name}' ...")
|
|
if self.mode == "systemd":
|
|
result = self._run("systemctl", "stop", self.name)
|
|
else:
|
|
result = self._run(f"/etc/init.d/{self.name}", "stop")
|
|
if result.returncode != 0:
|
|
raise AbortedError(
|
|
f"Dienst konnte nicht gestoppt werden: {result.stderr.strip() or result.stdout.strip()}"
|
|
)
|
|
|
|
# Warten, bis der Dienst wirklich unten ist - sonst schreibt Kerio
|
|
# noch in Dateien, die wir gerade verschieben.
|
|
deadline = time.monotonic() + self.wait_seconds
|
|
while time.monotonic() < deadline:
|
|
if self.is_running() is False:
|
|
print(" Dienst gestoppt.")
|
|
return
|
|
time.sleep(1)
|
|
raise AbortedError(
|
|
f"Dienst '{self.name}' laeuft nach {self.wait_seconds}s immer noch. Abbruch."
|
|
)
|
|
|
|
def start(self) -> None:
|
|
if self.mode == "none":
|
|
return
|
|
if self.dry_run:
|
|
print(f" [dry-run] wuerde Dienst '{self.name}' starten")
|
|
return
|
|
|
|
print(f" Starte Dienst '{self.name}' ...")
|
|
if self.mode == "systemd":
|
|
result = self._run("systemctl", "start", self.name)
|
|
else:
|
|
result = self._run(f"/etc/init.d/{self.name}", "start")
|
|
if result.returncode != 0:
|
|
print(f" WARNUNG: Start fehlgeschlagen: "
|
|
f"{result.stderr.strip() or result.stdout.strip()}", file=sys.stderr)
|
|
else:
|
|
print(" Dienst gestartet.")
|
|
|
|
|
|
# ==========================================================================
|
|
# Teil 3: 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)
|
|
|
|
|
|
# ==========================================================================
|
|
# Teil 4: 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:]))
|