Hacker-Software/kerio-connect-rename-user
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Umbenennung ueber users.cfg statt Administration API

Browse Source 
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>
This commit is contained in:
duffyduck
2026-08-07 13:44:31 +02:00
co-authored by Claude Opus 5
parent b12041fc81
commit 5011491067
5 changed files with 715 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+92 -74
View File
@@ -6,62 +6,75 @@ Werkzeuge für Kerio Connect auf der Kommandozeile.
| --- | --- | | --- | --- |
| [`kerio_rename_user.py`](kerio_rename_user.py) | Benutzer umbenennen — Login-Name **und** Mailbox-Verzeichnis im Store | | [`kerio_rename_user.py`](kerio_rename_user.py) | Benutzer umbenennen — Login-Name **und** Mailbox-Verzeichnis im Store |
| [`kerio_trial_setup.py`](kerio_trial_setup.py) | Am Erstkonfigurationsassistenten vorbei, wenn dessen Lizenzschritt klemmt | | [`kerio_trial_setup.py`](kerio_trial_setup.py) | Am Erstkonfigurationsassistenten vorbei, wenn dessen Lizenzschritt klemmt |
| [`kerio_common.py`](kerio_common.py) | Gemeinsame Bausteine: API-Client, Dienststeuerung | | [`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 |
Beide Scripts laufen **lokal auf dem Kerio-Server als root**, brauchen nur die Beide Scripts laufen **lokal auf dem Kerio-Server als root**, brauchen nur die
Python-Standardbibliothek und bieten `--dry-run`. Python-Standardbibliothek und bieten `--dry-run`.
Getestet gegen **Kerio Connect 10.0.9 patch 2** auf Debian 12.
--- ---
# Benutzer umbenennen # Benutzer umbenennen
## Warum ## Warum das nicht trivial ist
Kerio Connect legt die Mailbox eines Benutzers unter Kerio hält den Benutzernamen an zwei Stellen, die nichts voneinander wissen:
``` | Wo | Was |
<store>/mail/<domain>/<loginname>/ | --- | --- |
``` | `users.cfg` | Der Login-Name, in **zwei** Listen: `User` und `UserAdditionalData` |
| Dateisystem | `<store>/mail/<domain>/<loginname>/` — die eigentliche Mailbox |
ab. Ändert man den Login-Namen über die Administration API oder die Ändert man nur den Login-Namen, sucht Kerio ab sofort im Verzeichnis mit dem
Admin-Konsole, zieht dieses Verzeichnis nicht mit — der Benutzer findet danach *neuen* Namen, findet keins, legt ein leeres an — und die alten Mails liegen
eine leere Mailbox vor, während die alten Daten unter dem alten Verzeichnisnamen weiter unter dem alten Verzeichnisnamen. Typischer Anlass für eine Umbenennung:
liegenbleiben. Typischer Anlass: Heirat oder ein anderer Namenswechsel. Heirat oder ein anderer Namenswechsel.
Dieses Script erledigt beide Hälften in der richtigen Reihenfolge. ## Warum nicht über die Administration API
## Voraussetzungen **Die API kann das nicht.** `Users.set` nimmt ein Feld `loginName` entgegen und
ignoriert es stillschweigend — kein Fehler, keine Wirkung. Nachgeprüft gegen
Kerio Connect 10.0.9 mit vier Varianten (nur `loginName`, mit `domainId`, mit
leeren `emailAddresses`, vollständiges User-Objekt): alle wirkungslos, während
`fullName` im selben Aufruf sauber übernommen wird.
- Python 3.9 oder neuer, nur Standardbibliothek — keine Pakete zu installieren Deshalb arbeitet das Script direkt auf `users.cfg` — und braucht dadurch
- Läuft **lokal auf dem Kerio-Server als root** (Dateisystemzugriff + Dienstkontrolle) **keine Admin-Zugangsdaten**.
- Administration API erreichbar (Standard: `https://localhost:4040/admin/api/jsonrpc/`)
## Ablauf ## Ablauf
1. **Vorprüfungen**, bevor irgendetwas geschrieben wird: 1. **Vorprüfungen**, bevor irgendetwas geschrieben wird:
- Domain und Benutzer über die API auflösen - Domain und Benutzer in `users.cfg` vorhanden?
- Benutzer aus einem Verzeichnisdienst (LDAP/AD)? → Abbruch, denn der Name - Benutzer aus einem Verzeichnisdienst (`InternalDb=0`)? → Abbruch, denn
gehört dann dem Verzeichnisdienst dann gehört der Name dem Verzeichnisdienst
- neuer Login-Name noch frei? - neuer Login-Name frei und zeichenmäßig zulässig?
- Mailbox-Verzeichnis vorhanden, Zielname noch nicht belegt, - Zielverzeichnis unbelegt, Quelle und Ziel auf demselben Dateisystem?
Quelle und Ziel auf demselben Dateisystem? - Dienst steuerbar?
- ist der Dienst überhaupt steuerbar?
2. Plan ausgeben. `--dry-run` endet hier, sonst Rückfrage (`--yes` überspringt). 2. Plan ausgeben. `--dry-run` endet hier, sonst Rückfrage (`--yes` überspringt).
3. `Users.set` setzt `loginName` und optional `fullName`. 3. Dienst stoppen.
4. Kerio-Dienst stoppen und warten, bis er wirklich unten ist. 4. `users.cfg` **neu einlesen**, Name in `User` und `UserAdditionalData` setzen,
5. Verzeichnisse per `os.rename` verschieben (`mail/`, und `archive/` falls vorhanden). atomar über eine temporäre Datei speichern.
6. Dienst wieder starten. 5. Verzeichnisse verschieben (`mail/`, und `archive/` falls vorhanden).
6. Dienst starten und gegenlesen.
Schlägt Schritt 5 fehl, werden bereits verschobene Verzeichnisse zurückgenommen **Schritte 4 und 5 liegen bewusst in einem einzigen Stopp-Fenster.** Das ist
und der Login-Name in Kerio auf den alten Wert zurückgesetzt — Konfiguration und keine Bequemlichkeit: Läuft Kerio zwischendurch mit nur einer der beiden
Platte bleiben konsistent. Hälften, legt es die vermeintlich fehlende Mailbox sofort neu an, und man hat
ein verwaistes Verzeichnis mehr. Genau das ist beim Entwickeln passiert.
Schlägt etwas fehl, werden verschobene Verzeichnisse zurückgenommen und
`users.cfg` aus dem Backup wiederhergestellt.
## Benutzung ## Benutzung
```bash ```bash
# Erst ansehen, was passieren würde (braucht kein root): # Wer ist da? (kein root nötig)
./kerio_rename_user.py --list
# Erst ansehen:
./kerio_rename_user.py \ ./kerio_rename_user.py \
--admin admin \
--domain firma.de \ --domain firma.de \
--old-login anna.mueller \ --old-login anna.mueller \
--new-login anna.schmidt \ --new-login anna.schmidt \
@@ -69,22 +82,17 @@ Platte bleiben konsistent.
--dry-run --dry-run
# Dann wirklich: # Dann wirklich:
sudo -E ./kerio_rename_user.py \ sudo ./kerio_rename_user.py \
--admin admin \
--domain firma.de \ --domain firma.de \
--old-login anna.mueller \ --old-login anna.mueller \
--new-login anna.schmidt \ --new-login anna.schmidt \
--full-name "Anna Schmidt" --full-name "Anna Schmidt"
``` ```
Das Passwort kommt aus der Umgebungsvariablen `KERIO_ADMIN_PASSWORD` oder wird
interaktiv abgefragt — nicht über die Kommandozeile, sonst landet es in der
Shell-History. (`sudo -E` reicht die Variable durch.)
Nur den angezeigten Namen ändern, ohne Login und Verzeichnis anzufassen: Nur den angezeigten Namen ändern, ohne Login und Verzeichnis anzufassen:
```bash ```bash
./kerio_rename_user.py --admin admin --domain firma.de \ sudo ./kerio_rename_user.py --domain firma.de \
--old-login anna.mueller --full-name "Anna Schmidt" --old-login anna.mueller --full-name "Anna Schmidt"
``` ```
@@ -92,35 +100,50 @@ Nur den angezeigten Namen ändern, ohne Login und Verzeichnis anzufassen:
| Option | Bedeutung | | Option | Bedeutung |
| --- | --- | | --- | --- |
| `--server`, `--port` | Admin-API, Standard `localhost:4040` | | `--list` | Benutzer aus `users.cfg` auflisten |
| `--admin`, `--password` | Admin-Zugang; besser `KERIO_ADMIN_PASSWORD` |
| `--domain` | Mail-Domain, z. B. `firma.de` | | `--domain` | Mail-Domain, z. B. `firma.de` |
| `--old-login`, `--new-login` | Login-Namen ohne `@domain` | | `--old-login`, `--new-login` | Login-Namen ohne `@domain` |
| `--full-name` | Neuer angezeigter Name | | `--full-name` | Neuer angezeigter Name |
| `--store-dir` | Store-Pfad, Standard `/opt/kerio/mailserver/store` | | `--install-dir` | Standard `/opt/kerio/mailserver` |
| `--store-dir` | Standard `/opt/kerio/mailserver/store` |
| `--service-name` | Dienstname, Standard `kerio-connect` | | `--service-name` | Dienstname, Standard `kerio-connect` |
| `--service-manager` | `auto`, `systemd`, `initd` oder `none` | | `--service-manager` | `auto`, `systemd`, `initd` oder `none` |
| `--service-wait` | Sekunden Wartezeit aufs Stoppen, Standard 90 | | `--service-wait` | Sekunden Wartezeit aufs Stoppen, Standard 90 |
| `--dry-run` | Nur anzeigen | | `--dry-run` | Nur anzeigen |
| `--yes` | Rückfrage überspringen | | `--yes` | Rückfrage überspringen |
| `--force` | Auch bei LDAP-/AD-Benutzern versuchen | | `--force` | Auch bei Benutzern aus einem Verzeichnisdienst |
| `--insecure` | TLS-Zertifikat nicht prüfen (selbstsigniert) |
`--service-manager none` fasst den Dienst nicht an — dafür muss Kerio dann `--service-manager none` fasst den Dienst nicht an — dafür muss Kerio dann
selbst gestoppt sein, bevor das Script läuft. selbst gestoppt sein, bevor das Script läuft.
## Was das Script bewusst nicht anfasst
Nach der Umbenennung steht der alte Name noch an ein paar Stellen, die reine
Anzeige betreffen und den Mailbetrieb nicht stören:
- `.personal` — die vCard des Benutzers, enthält noch die alte Adresse
- `#public/Contacts/` — der Eintrag in der globalen Adressliste
- `.caldav.db` / `.carddav.db` — die DAV-Datenbanken
Kerio schreibt `.personal` beim Start teilweise selbst neu. Wer die Adresse dort
sofort korrekt haben will, korrigiert den Kontakt in der Administrationskonsole.
Kommt der alte Name in `users.cfg` außerhalb der Benutzerlisten vor (etwa in
einem Alias), meldet das Script das im Plan und ändert es **nicht** automatisch —
lieber ein Hinweis als eine kaputte Konfiguration.
Die alte Adresse nimmt nach der Umbenennung keine Mail mehr an. Wer das braucht,
legt anschließend in der Administrationskonsole einen Alias an.
## Hinweise ## Hinweise
- **Vorher ein Backup des Stores anlegen.** Das Script verschiebt Verzeichnisse. - **Vorher ein Backup des Stores anlegen.** Das Script verschiebt Verzeichnisse.
- Erst an einem Testbenutzer ausprobieren. - Erst an einem Testbenutzer ausprobieren.
- Der Benutzer sollte während des Laufs nicht angemeldet sein. - Der Benutzer sollte während des Laufs nicht angemeldet sein.
- Nach der Umbenennung müssen sich alle Mail-Clients des Benutzers mit dem neuen - Nach der Umbenennung müssen sich alle Mail-Clients mit dem neuen Login neu
Login neu anmelden. anmelden.
- Die alte Adresse wird bewusst **nicht** als Alias angelegt. Wer möchte, dass - Gruppenmitgliedschaften bleiben erhalten: Kerio referenziert Gruppen über
Mail an die alte Adresse weiterhin ankommt, legt den Alias anschließend in der GUIDs, nicht über Namen.
Admin-Konsole an.
- Weitere Adressen am Benutzer (`emailAddresses`) bleiben unverändert und werden
im Plan zur Kontrolle mit angezeigt.
--- ---
@@ -135,13 +158,22 @@ verlinkte Registrierungsseite ist seit der Übernahme durch GFI teilweise tot
man kommt im Assistenten nicht weiter. man kommt im Assistenten nicht weiter.
Kerio kennt dafür den Schalter `ConfigWizardDone` in `mailserver.cfg`. Steht er Kerio kennt dafür den Schalter `ConfigWizardDone` in `mailserver.cfg`. Steht er
auf `1`, öffnet die Administrationskonsole den Assistenten nicht mehr und der auf `1`, öffnet die Administrationskonsole den Assistenten nicht mehr.
Server läuft 30 Tage ab Installation im unregistrierten Modus.
Im unregistrierten Modus fehlen Antivirus-Updates, ActiveSync, Greylisting und ## Was das bringt — und was nicht
Hersteller-Support; nach 30 Tagen stoppt die Engine. **Benutzerverwaltung,
Administrationskonsole und Admin-API sind nicht eingeschränkt** — für einen Danach sind **Administrationskonsole und Admin-API voll nutzbar**, und dieses
Testserver also unproblematisch. Repo lässt sich damit testen.
**Der Mailbetrieb bleibt aber gesperrt.** Ohne Lizenzschlüssel weisen IMAP, POP3
und Webmail jede Verbindung ab (`Server license expired` beim Client,
`Product name does not match` im Log), und `error.log` meldet beim Start
`No license key found`. Der Schalter überspringt nur den Dialog — er aktiviert
keinen Testzeitraum.
Für echten Mailbetrieb führt kein Weg an einer Trial-Lizenznummer vorbei:
**Dashboard → „Become a registered trial user"**, oder auf der GFI-Downloadseite
„Try free for 30 days". Braucht ausgehendes HTTPS auf Port 443.
## Benutzung ## Benutzung
@@ -151,33 +183,19 @@ sudo ./kerio_trial_setup.py --skip-wizard # Assistenten überspringen
sudo ./kerio_trial_setup.py --restore # Backup zurückspielen sudo ./kerio_trial_setup.py --restore # Backup zurückspielen
``` ```
`--status` zeigt Dienstzustand, `ConfigWizardDone`, `TrialID`, vorhandene
Lizenzdateien und eine Schätzung, wann der Testzeitraum abläuft.
## Der springende Punkt ## Der springende Punkt
`mailserver.cfg` darf **nur bei gestopptem Dienst** bearbeitet werden. Kerio `mailserver.cfg` darf **nur bei gestopptem Dienst** bearbeitet werden. Kerio
schreibt seine Konfiguration beim Beenden aus dem Speicher zurück und würde eine schreibt seine Konfiguration beim Beenden aus dem Speicher zurück und würde eine
Änderung am laufenden Dienst wieder überschreiben. Das Script stoppt deshalb den Änderung am laufenden Dienst wieder überschreiben. Das gilt für `users.cfg`
Dienst, liest die Datei **danach neu ein** (die Version von vorher ist zu diesem genauso — beide Scripts halten sich daran.
Zeitpunkt bereits veraltet), schreibt atomar über eine temporäre Datei im selben
Verzeichnis, startet den Dienst und liest den Wert zur Kontrolle noch einmal.
Vor jeder Änderung wird ein Backup als `mailserver.cfg.bak-trial-setup` angelegt.
## Falls du doch eine registrierte Trial willst
Geht aus der laufenden Instanz heraus, ohne Installer: **Dashboard → „Become a
registered trial user"**. Braucht ausgehendes HTTPS auf Port 443 zum
GFI-Registrierungsserver.
--- ---
## API- und Doku-Referenz ## Referenzen
- [Administration API for Kerio Connect — Users](https://manuals.gfi.com/en/kerio/api/connect/admin/reference/interfacekerio_1_1jsonapi_1_1admin_1_1_users.html) - [Administration API — Users](https://manuals.gfi.com/en/kerio/api/connect/admin/reference/interfacekerio_1_1jsonapi_1_1admin_1_1_users.html)
- [Sample communication (Session.login, X-Token)](https://manuals.gfi.com/en/kerio/api/connect/admin/reference/sample_communication.html) - [Sample communication (Session.login, X-Token)](https://manuals.gfi.com/en/kerio/api/connect/admin/reference/sample_communication.html)
- [User struct](https://manuals.gfi.com/en/kerio/api/connect/admin/reference/structkerio_1_1jsonapi_1_1admin_1_1_user.html)
- [Registering Kerio Connect (unregistered mode)](https://manuals.gfi.com/en/kerio/connect/content/registration-and-licenses/registering-kerio-connect-1134.html) - [Registering Kerio Connect (unregistered mode)](https://manuals.gfi.com/en/kerio/connect/content/registration-and-licenses/registering-kerio-connect-1134.html)
- [Initial Wizard Settings Are Not Saving (ConfigWizardDone)](https://support.kerioconnect.gfi.com/article/114858-initial-wizard-settings-are-not-saving-in-kerio-connect) - [Initial Wizard Settings Are Not Saving (ConfigWizardDone)](https://support.kerioconnect.gfi.com/article/114858-initial-wizard-settings-are-not-saving-in-kerio-connect)
- [Modifying the mailserver.cfg](https://support.kerioconnect.gfi.com/en-us/article/114788-modifying-the-mailserver-cfg) - [Modifying the mailserver.cfg](https://support.kerioconnect.gfi.com/en-us/article/114788-modifying-the-mailserver-cfg)
kerio_rename_user.py
+181 -161
View File
@@ -2,22 +2,29 @@
""" """
Benennt einen Benutzer in Kerio Connect vollstaendig um: Benennt einen Benutzer in Kerio Connect vollstaendig um:
1. Login-Name (und optional voller Name) ueber die Administration API 1. den Login-Namen in users.cfg (und optional den angezeigten Namen)
2. das zugehoerige Mailbox-Verzeichnis im Store auf der Platte 2. das zugehoerige Mailbox-Verzeichnis im Store auf der Platte
Kerio legt die Mailbox unter <store>/mail/<domain>/<loginname>/ ab und zieht Warum nicht ueber die Administration API: Kerio nimmt loginName in Users.set
das Verzeichnis beim Aendern des Login-Namens nicht mit. Ohne Schritt 2 findet zwar entgegen, ignoriert das Feld aber stillschweigend - kein Fehler, keine
der Benutzer nach der Umbenennung eine leere Mailbox vor. 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.
Damit das Verzeichnis gefahrlos verschoben werden kann, wird der Kerio-Dienst Kerio legt die Mailbox unter <store>/mail/<domain>/<loginname>/ ab. Ohne
waehrend des Umbenennens gestoppt und danach wieder gestartet. Schritt 2 findet der Benutzer nach der Umbenennung eine leere Mailbox vor,
waehrend die alten Mails unter dem alten Verzeichnisnamen liegenbleiben.
Das Script muss deshalb LOKAL AUF DEM KERIO-SERVER ALS ROOT laufen. 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.
Beispiel: Das Script muss LOKAL AUF DEM KERIO-SERVER ALS ROOT laufen.
Beispiele:
./kerio_rename_user.py --list # Benutzer anzeigen
sudo ./kerio_rename_user.py \\ sudo ./kerio_rename_user.py \\
--admin admin \\
--domain firma.de \\ --domain firma.de \\
--old-login anna.mueller \\ --old-login anna.mueller \\
--new-login anna.schmidt \\ --new-login anna.schmidt \\
@@ -30,28 +37,26 @@ Ohne --dry-run wird vor dem Schreiben nachgefragt (ausser mit --yes).
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import getpass
import json
import os import os
import shutil
import sys import sys
import time
from pathlib import Path from pathlib import Path
from kerio_common import ( from kerio_common import (
DEFAULT_INSTALL_DIR,
DEFAULT_SERVICE, DEFAULT_SERVICE,
DEFAULT_STORE_DIR, DEFAULT_STORE_DIR,
AbortedError, AbortedError,
KerioAdminApi,
KerioError,
ServiceController, ServiceController,
find_domain,
find_user,
) )
from kerio_users_config import UsersConfig, validate_login
# Store-Unterbaeume, in denen pro Benutzer ein Verzeichnis liegen kann. # Store-Unterbaeume, in denen pro Benutzer ein Verzeichnis liegen kann.
# Alle folgen dem Schema <store>/<subtree>/<domain>/<loginname>/ # Alle folgen dem Schema <store>/<subtree>/<domain>/<loginname>/
STORE_SUBTREES = ("mail", "archive") STORE_SUBTREES = ("mail", "archive")
BACKUP_SUFFIX = ".bak-rename-user"
# ========================================================================== # ==========================================================================
# Store-Verzeichnisse # Store-Verzeichnisse
@@ -97,13 +102,6 @@ def plan_store_renames(store_dir: Path, domain_name: str, old_login: str,
# Schreibweise des vorhandenen Verzeichnisses uebernehmen. # Schreibweise des vorhandenen Verzeichnisses uebernehmen.
target_name = new_login.lower() if source.name.islower() else new_login target_name = new_login.lower() if source.name.islower() else new_login
renames.append((source, domain_dir / target_name)) 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 return renames
@@ -126,19 +124,14 @@ def check_rename_targets(renames: list[tuple[Path, Path]]) -> None:
) )
def apply_store_renames(renames: list[tuple[Path, Path]], def apply_store_renames(renames: list[tuple[Path, Path]]) -> list[tuple[Path, Path]]:
dry_run: bool) -> list[tuple[Path, Path]]:
"""Fuehrt die Verschiebungen aus. Gibt zurueck, was tatsaechlich getan wurde.""" """Fuehrt die Verschiebungen aus. Gibt zurueck, was tatsaechlich getan wurde."""
done: list[tuple[Path, Path]] = [] done: list[tuple[Path, Path]] = []
for source, target in renames: for source, target in renames:
if dry_run:
print(f" [dry-run] wuerde verschieben: {source} -> {target}")
continue
try: try:
os.rename(source, target) os.rename(source, target)
except OSError as exc: except OSError as exc:
# Bereits erfolgte Verschiebungen zuruecknehmen. revert_store_renames(done)
_revert_store_renames(done)
raise AbortedError( raise AbortedError(
f"Verschieben von '{source}' nach '{target}' fehlgeschlagen: {exc}" f"Verschieben von '{source}' nach '{target}' fehlgeschlagen: {exc}"
) from exc ) from exc
@@ -147,7 +140,7 @@ def apply_store_renames(renames: list[tuple[Path, Path]],
return done return done
def _revert_store_renames(done: list[tuple[Path, Path]]) -> None: def revert_store_renames(done: list[tuple[Path, Path]]) -> None:
for source, target in reversed(done): for source, target in reversed(done):
try: try:
os.rename(target, source) os.rename(target, source)
@@ -162,82 +155,105 @@ def _revert_store_renames(done: list[tuple[Path, Path]]) -> None:
# ========================================================================== # ==========================================================================
def rename_user(api: KerioAdminApi, args: argparse.Namespace, 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: service: ServiceController) -> int:
store_dir = Path(args.store_dir) store_dir = Path(args.store_dir)
# -- Vorpruefungen: nichts wird geschrieben --------------------------- # -- Vorpruefungen: nichts wird geschrieben ---------------------------
domain = find_domain(api, args.domain) domains = config.domains()
domain_id = domain["id"] domain = next((d for d in domains if d.lower() == args.domain.lower()), None)
domain_name = domain["name"] if domain is None:
print(f"Domain: {domain_name} (id={domain_id})") raise AbortedError(
f"Domain '{args.domain}' kommt in users.cfg nicht vor. "
f"Vorhanden: {', '.join(domains) or '(keine)'}"
)
user = find_user(api, domain_id, args.old_login) user = config.find_user(args.old_login, domain)
if user is None: if user is None:
raise KerioError( raise AbortedError(
f"Benutzer '{args.old_login}' existiert nicht in Domain '{domain_name}'." f"Benutzer '{args.old_login}' existiert nicht in Domain '{domain}'."
) )
old_login = user["loginName"] old_login = user["loginName"]
new_login = args.new_login or old_login 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: if new_login != old_login:
pattern["loginName"] = new_login validate_login(new_login)
if args.full_name and args.full_name != user.get("fullName"):
pattern["fullName"] = args.full_name 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]] = [] renames: list[tuple[Path, Path]] = []
if new_login != old_login: if new_login != old_login:
renames = plan_store_renames(store_dir, domain_name, old_login, new_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) check_rename_targets(renames)
service.preflight() service.preflight()
if not pattern and not renames: 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.") print("\nNichts zu tun.")
return 0 return 0
# -- Plan ausgeben ---------------------------------------------------- # -- Plan ausgeben ----------------------------------------------------
print("\nGeplante Aenderungen") print("\nGeplante Aenderungen")
print("--------------------") print("--------------------")
print(f" Benutzer-ID : {user['id']}") print(f" Domain : {domain}")
print(f" Guid : {user['guid']}")
if "loginName" in pattern: if needs_name:
print(f" Login-Name : {old_login} -> {new_login}") print(f" Login-Name : {old_login} -> {new_login}")
print(f" Adresse : {old_login}@{domain} -> {new_login}@{domain}")
else: else:
print(f" Login-Name : {old_login} (unveraendert)") print(f" Login-Name : {old_login} (unveraendert)")
old_full_name = user.get("fullName", "") if needs_fullname:
if "fullName" in pattern: print(f" Voller Name : {user['fullName']} -> {args.full_name}")
print(f" Voller Name : {old_full_name} -> {args.full_name}")
else: else:
print(f" Voller Name : {old_full_name} (unveraendert)") print(f" Voller Name : {user['fullName']} (unveraendert)")
print(f" Adresse : {old_login}@{domain_name} -> {new_login}@{domain_name}")
print(f" users.cfg : {config.path}")
if renames: if renames:
print(f" Dienst : {service.describe()}") print(f" Dienst : {service.describe()}")
for source, target in renames: for source, target in renames:
print(f" Verzeichnis : {source}") print(f" Verzeichnis : {source}")
print(f" -> {target}") print(f" -> {target}")
addresses = user.get("emailAddresses") or [] extra = config.other_occurrences(old_login, domain) if needs_name else []
if addresses: if extra:
print(f" Weitere Adressen am Benutzer (unveraendert): " print("\n ACHTUNG: Der alte Name kommt auch hier vor und wird NICHT "
f"{json.dumps(addresses, ensure_ascii=False)}") "automatisch geaendert:")
for hit in extra:
print(f" - {hit}")
print(" Bitte nach der Umbenennung in der Administrationskonsole pruefen.")
if args.dry_run: if args.dry_run:
print("\n--dry-run: es wurde nichts geaendert.") print("\n--dry-run: es wurde nichts geaendert.")
@@ -245,74 +261,78 @@ def rename_user(api: KerioAdminApi, args: argparse.Namespace,
if not args.yes: if not args.yes:
print("\nAchtung: der Kerio-Dienst wird dabei kurz gestoppt.") print("\nAchtung: der Kerio-Dienst wird dabei kurz gestoppt.")
answer = input("Aenderungen jetzt durchfuehren? [j/N] ").strip().lower() if input("Aenderungen jetzt durchfuehren? [j/N] ").strip().lower() \
if answer not in ("j", "ja", "y", "yes"): not in ("j", "ja", "y", "yes"):
print("Abgebrochen.") print("Abgebrochen.")
return 1 return 1
# -- Schreiben -------------------------------------------------------- # -- Schreiben: users.cfg und Verzeichnisse in einem Stopp-Fenster ----
print() print()
if pattern: was_running = service.is_running()
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() service.stop()
# Nach dem Stoppen erneut pruefen: Kerio koennte in der Zwischenzeit
# ein leeres Verzeichnis unter dem neuen Namen angelegt haben. 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) check_rename_targets(renames)
apply_store_renames(renames, dry_run=False) done = apply_store_renames(renames)
except AbortedError as exc: except (AbortedError, OSError) as exc:
print(f"\nFEHLER beim Verschieben: {exc}", file=sys.stderr) print(f"\nFEHLER: {exc}", file=sys.stderr)
_rollback_api_change(args, user, old_login, service) 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 return 2
finally:
if was_running:
service.start() service.start()
print(f"\nFertig. Benutzer heisst jetzt '{new_login}', " # -- Gegenlesen -------------------------------------------------------
f"Mailbox liegt unter '{renames[0][1]}'.") 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 " print("Hinweis: Der Benutzer muss sich in allen Mail-Clients mit dem "
"neuen Login neu anmelden.") "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 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 # CLI
# ========================================================================== # ==========================================================================
@@ -320,67 +340,64 @@ def _rollback_api_change(args: argparse.Namespace, user: dict, old_login: str,
def parse_args(argv: list[str]) -> argparse.Namespace: def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description="Benennt einen Kerio-Connect-Benutzer inklusive Store-Verzeichnis um.", description="Benennt einen Kerio-Connect-Benutzer inklusive "
"Store-Verzeichnis um.",
formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=__doc__, epilog=__doc__,
) )
parser.add_argument("--server", default="localhost", parser.add_argument("--install-dir", default=DEFAULT_INSTALL_DIR,
help="Hostname der Admin-API (Standard: localhost)") help=f"Kerio-Installationsverzeichnis "
parser.add_argument("--port", type=int, default=4040, f"(Standard: {DEFAULT_INSTALL_DIR})")
help="Port der Admin-API (Standard: 4040)") parser.add_argument("--store-dir", default=DEFAULT_STORE_DIR,
parser.add_argument("--admin", required=True, help=f"Kerio-Store-Verzeichnis (Standard: {DEFAULT_STORE_DIR})")
help="Benutzername des Administrators") parser.add_argument("--domain",
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") help="Mail-Domain des Benutzers, z. B. firma.de")
parser.add_argument("--old-login", required=True, parser.add_argument("--old-login",
help="Bisheriger Login-Name (ohne @domain)") help="Bisheriger Login-Name (ohne @domain)")
parser.add_argument("--new-login", parser.add_argument("--new-login",
help="Neuer Login-Name (ohne @domain)") help="Neuer Login-Name (ohne @domain)")
parser.add_argument("--full-name", parser.add_argument("--full-name",
help='Neuer voller Name, z. B. "Anna Schmidt"') help='Neuer angezeigter Name, z. B. "Anna Schmidt"')
parser.add_argument("--store-dir", default=DEFAULT_STORE_DIR, parser.add_argument("--list", action="store_true",
help=f"Kerio-Store-Verzeichnis (Standard: {DEFAULT_STORE_DIR})") help="Benutzer aus users.cfg auflisten und beenden")
parser.add_argument("--service-name", default=DEFAULT_SERVICE, parser.add_argument("--service-name", default=DEFAULT_SERVICE,
help=f"Name des Kerio-Dienstes (Standard: {DEFAULT_SERVICE})") help=f"Name des Kerio-Dienstes (Standard: {DEFAULT_SERVICE})")
parser.add_argument("--service-manager", default="auto", parser.add_argument("--service-manager", default="auto",
choices=("auto", "systemd", "initd", "none"), choices=("auto", "systemd", "initd", "none"),
help="Wie der Dienst gestoppt wird. 'none' fasst ihn nicht an - " help="Wie der Dienst gestoppt wird. 'none' fasst ihn nicht "
"dann muss Kerio selbst gestoppt werden.") "an - dann muss Kerio selbst gestoppt sein.")
parser.add_argument("--service-wait", type=int, default=90, parser.add_argument("--service-wait", type=int, default=90,
help="Sekunden, die auf das Stoppen gewartet wird (Standard: 90)") help="Sekunden Wartezeit aufs Stoppen (Standard: 90)")
parser.add_argument("--dry-run", action="store_true", parser.add_argument("--dry-run", action="store_true",
help="Nur anzeigen, was passieren wuerde") help="Nur anzeigen, was passieren wuerde")
parser.add_argument("--yes", action="store_true", parser.add_argument("--yes", action="store_true",
help="Rueckfrage ueberspringen") help="Rueckfrage ueberspringen")
parser.add_argument("--force", action="store_true", parser.add_argument("--force", action="store_true",
help="Auch bei LDAP-/AD-gemappten Benutzern versuchen") help="Auch bei Benutzern aus einem Verzeichnisdienst")
parser.add_argument("--insecure", action="store_true",
help="TLS-Zertifikat nicht pruefen (selbstsigniertes Zertifikat)")
args = parser.parse_args(argv) 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: if not args.new_login and not args.full_name:
parser.error("Mindestens --new-login oder --full-name angeben.") parser.error("Mindestens --new-login oder --full-name angeben.")
if not args.dry_run and os.geteuid() != 0:
if args.new_login and not args.dry_run and os.geteuid() != 0:
parser.error( parser.error(
"Zum Verschieben des Store-Verzeichnisses werden root-Rechte benoetigt. " "Zum Aendern von users.cfg und zum Verschieben des "
"Store-Verzeichnisses werden root-Rechte benoetigt. "
"Mit sudo starten (oder --dry-run zum Testen)." "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 return args
def main(argv: list[str]) -> int: def main(argv: list[str]) -> int:
args = parse_args(argv) args = parse_args(argv)
config_path = Path(args.install_dir) / "users.cfg"
service = ServiceController( service = ServiceController(
name=args.service_name, name=args.service_name,
@@ -390,14 +407,17 @@ def main(argv: list[str]) -> int:
) )
try: try:
with KerioAdminApi(args.server, args.port, config = UsersConfig(config_path)
verify_tls=not args.insecure) as api: if args.list:
api.login(args.admin, args.password) return list_users(config, args.domain)
print(f"Angemeldet an {args.server}:{args.port} als {args.admin}") return rename_user(args, config, service)
return rename_user(api, args, service) except AbortedError as exc:
except (KerioError, AbortedError) as exc:
print(f"\nFEHLER: {exc}", file=sys.stderr) print(f"\nFEHLER: {exc}", file=sys.stderr)
return 2 return 2
except PermissionError as exc:
print(f"\nFEHLER: Keine Berechtigung: {exc}\nMit sudo starten.",
file=sys.stderr)
return 2
except KeyboardInterrupt: except KeyboardInterrupt:
print("\nAbgebrochen.", file=sys.stderr) print("\nAbgebrochen.", file=sys.stderr)
return 130 return 130
kerio_users_config.py
+209
View File
@@ -0,0 +1,209 @@
"""
Liest und aendert /opt/kerio/mailserver/users.cfg.
Warum diese Datei und nicht 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 dieser Datei aendern.
Aufbau der Datei (XML, UTF-8):
<config>
<list name="User">
<listitem>
<variable name="Name">max.mustermann</variable>
<variable name="Domain">local.lan</variable>
<variable name="Guid">94480090-...</variable>
<variable name="Fullname">Max Mustermann</variable>
...
</listitem>
</list>
<list name="Group"> ... </list>
<list name="Alias"> ... </list>
<list name="UserAdditionalData">
<listitem>
<variable name="Name">max.mustermann</variable>
<variable name="Domain">local.lan</variable>
<variable name="InternalDb">1</variable>
...
Der Login-Name steht in ZWEI Listen: 'User' und 'UserAdditionalData'. Beide
verweisen ueber Name+Domain, nicht ueber die Guid - beide muessen mitgeaendert
werden, sonst verliert der Benutzer seine Passworthistorie.
Geaendert wird der Rohtext, nicht ein neu serialisierter XML-Baum: so bleiben
Formatierung und alle Felder, die wir nicht kennen, unangetastet.
"""
from __future__ import annotations
import os
import re
import shutil
from pathlib import Path
from kerio_common import AbortedError
# Listen, in denen ein Benutzer ueber <variable name="Name"> gefuehrt wird.
USER_LISTS = ("User", "UserAdditionalData")
# Erlaubte Zeichen fuer einen Login-Namen. Bewusst eng gehalten: der Name
# landet im Dateisystem als Verzeichnisname und im XML als Elementinhalt.
VALID_LOGIN = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9._-]{0,62}[A-Za-z0-9])?$")
_LIST_RE = re.compile(
r'<list name="(?P<name>[^"]+)">(?P<body>.*?)</list>', re.DOTALL)
_ITEM_RE = re.compile(r"<listitem>.*?</listitem>", re.DOTALL)
def _var(block: str, name: str) -> str | None:
match = re.search(rf'<variable name="{re.escape(name)}">([^<]*)</variable>', block)
return match.group(1) if match else None
def validate_login(name: str) -> None:
if not VALID_LOGIN.match(name):
raise AbortedError(
f"'{name}' ist als Login-Name nicht zulaessig. Erlaubt sind "
"Buchstaben, Ziffern, Punkt, Bindestrich und Unterstrich; "
"Anfang und Ende muessen alphanumerisch sein."
)
class UsersConfig:
"""Rohtext-Sicht auf users.cfg mit gezielten Aenderungen."""
def __init__(self, path: Path) -> None:
self.path = path
if not path.is_file():
raise AbortedError(
f"'{path}' existiert nicht. Installationsverzeichnis mit "
"--install-dir angeben."
)
self.text = path.read_text(encoding="utf-8")
# -- Lesen -------------------------------------------------------------
def items(self, list_name: str) -> list[str]:
"""Alle <listitem>-Bloecke einer Liste als Rohtext."""
for match in _LIST_RE.finditer(self.text):
if match.group("name") == list_name:
return _ITEM_RE.findall(match.group("body"))
return []
def domains(self) -> list[str]:
seen = []
for block in self.items("User"):
domain = _var(block, "Domain")
if domain and domain not in seen:
seen.append(domain)
return seen
def users(self, domain: str | None = None) -> list[dict]:
"""Benutzer als Dicts mit den Feldern, die uns interessieren."""
result = []
for block in self.items("User"):
entry = {
"loginName": _var(block, "Name") or "",
"domain": _var(block, "Domain") or "",
"fullName": _var(block, "Fullname") or "",
"guid": _var(block, "Guid") or "",
"enabled": _var(block, "Account_enabled") == "1",
}
if domain is None or entry["domain"].lower() == domain.lower():
result.append(entry)
return result
def find_user(self, login_name: str, domain: str) -> dict | None:
wanted = login_name.strip().lower()
for user in self.users(domain):
if user["loginName"].lower() == wanted:
return user
return None
def is_internal(self, login_name: str, domain: str) -> bool:
"""InternalDb=1 heisst: Benutzer liegt in Kerios eigener Datenbank.
Bei Benutzern aus einem Verzeichnisdienst gehoert der Name dorthin
und darf hier nicht geaendert werden.
"""
for block in self.items("UserAdditionalData"):
if (_var(block, "Name") or "").lower() == login_name.lower() and \
(_var(block, "Domain") or "").lower() == domain.lower():
internal = _var(block, "InternalDb")
return internal is None or internal == "1"
# Kein Eintrag: nicht raten, als intern behandeln.
return True
def other_occurrences(self, login_name: str, domain: str) -> list[str]:
"""Sucht den Namen ausserhalb von User/UserAdditionalData.
Kerio speichert Gruppenmitgliedschaften ueber Guids, daher ist hier
normalerweise nichts zu finden. Falls doch, wird gewarnt statt blind
zu aendern - besser ein Hinweis als eine kaputte Konfiguration.
"""
hits = []
for match in _LIST_RE.finditer(self.text):
list_name = match.group("name")
if list_name in USER_LISTS:
continue
for block in _ITEM_RE.findall(match.group("body")):
if re.search(rf"\b{re.escape(login_name)}\b", block):
label = _var(block, "Name") or "(ohne Name)"
hits.append(f"{list_name}: {label}")
return hits
# -- Schreiben ---------------------------------------------------------
def rename_user(self, old_login: str, new_login: str, domain: str,
full_name: str | None = None) -> int:
"""Setzt Name (und optional Fullname) in allen zustaendigen Listen.
Gibt die Anzahl geaenderter listitem-Bloecke zurueck.
"""
validate_login(new_login)
changed = 0
def patch_list(match: re.Match) -> str:
nonlocal changed
if match.group("name") not in USER_LISTS:
return match.group(0)
def patch_item(item: re.Match) -> str:
nonlocal changed
block = item.group(0)
if (_var(block, "Name") or "").lower() != old_login.lower():
return block
if (_var(block, "Domain") or "").lower() != domain.lower():
return block
new_block = re.sub(
r'(<variable name="Name">)[^<]*(</variable>)',
lambda m: m.group(1) + new_login + m.group(2),
block, count=1)
if full_name is not None and _var(block, "Fullname") is not None:
new_block = re.sub(
r'(<variable name="Fullname">)[^<]*(</variable>)',
lambda m: m.group(1) + full_name + m.group(2),
new_block, count=1)
changed += 1
return new_block
body = _ITEM_RE.sub(patch_item, match.group("body"))
return f'<list name="{match.group("name")}">{body}</list>'
self.text = _LIST_RE.sub(patch_list, self.text)
return changed
def backup(self, suffix: str) -> Path:
target = self.path.with_name(self.path.name + suffix)
shutil.copy2(self.path, target)
return target
def save(self) -> None:
# Ueber eine temporaere Datei im selben Verzeichnis, damit die
# Konfiguration nie halb geschrieben auf der Platte liegt.
temp = self.path.with_name(self.path.name + ".tmp-rename")
temp.write_text(self.text, encoding="utf-8")
shutil.copystat(self.path, temp)
os.replace(temp, self.path)
tests/test_store.py
+79
View File
@@ -0,0 +1,79 @@
"""Testet die Store-Logik gegen einen nachgebauten Kerio-Store."""
import sys, tempfile, os
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from kerio_rename_user import (
plan_store_renames, check_rename_targets, apply_store_renames,
find_dir_ci, AbortedError, revert_store_renames,
)
root = Path(tempfile.mkdtemp())
store = root / "store"
# Kerio-Store nachbauen: mail/ und archive/, Verzeichnisse klein geschrieben
for sub in ("mail", "archive"):
d = store / sub / "firma.de" / "anna.mueller"
d.mkdir(parents=True)
(d / "INBOX").mkdir()
(d / "INBOX" / "msg.eml").write_text("hallo")
# ein anderer Benutzer, der nicht angefasst werden darf
(store / "mail" / "firma.de" / "bob").mkdir(parents=True)
print("== 1. Planung ==")
renames = plan_store_renames(store, "firma.de", "anna.mueller", "anna.schmidt")
for s, t in renames:
print(f" {s.relative_to(store)} -> {t.relative_to(store)}")
assert len(renames) == 2, renames
check_rename_targets(renames)
print("\n== 2. Domain case-insensitiv (FIRMA.DE) ==")
r2 = plan_store_renames(store, "FIRMA.DE", "Anna.Mueller", "anna.schmidt")
assert len(r2) == 2, r2
print(" ok, findet trotz Gross-/Kleinschreibung")
print("\n== 3. Zielverzeichnis belegt -> Abbruch ==")
(store / "mail" / "firma.de" / "anna.schmidt").mkdir()
try:
check_rename_targets(renames)
print(" FEHLER: haette abbrechen muessen")
sys.exit(1)
except AbortedError as e:
print(f" ok: {e}")
(store / "mail" / "firma.de" / "anna.schmidt").rmdir()
print("\n== 4. Benutzer ohne Verzeichnis -> leere Liste, kein Abbruch ==")
r4 = plan_store_renames(store, "firma.de", "gibtsnicht", "neu")
assert r4 == [], r4
print(" ok, leere Liste (users.cfg wird dann trotzdem geaendert)")
print("\n== 5. (dry-run wird jetzt im Aufrufer behandelt) ==")
pass # dry-run liegt jetzt im Aufrufer
assert (store / "mail" / "firma.de" / "anna.mueller").is_dir()
print(" ok, Quelle noch da")
print("\n== 6. Tatsaechliches Verschieben ==")
done = apply_store_renames(renames)
assert len(done) == 2
for sub in ("mail", "archive"):
neu = store / sub / "firma.de" / "anna.schmidt"
assert neu.is_dir(), neu
assert (neu / "INBOX" / "msg.eml").read_text() == "hallo"
assert not (store / sub / "firma.de" / "anna.mueller").exists()
assert (store / "mail" / "firma.de" / "bob").is_dir(), "anderer Benutzer angefasst!"
print(" ok, Inhalte erhalten, Fremdbenutzer unberuehrt")
print("\n== 7. Rollback ==")
revert_store_renames(done)
for sub in ("mail", "archive"):
assert (store / sub / "firma.de" / "anna.mueller" / "INBOX" / "msg.eml").read_text() == "hallo"
print(" ok, alles zurueck")
print("\n== 8. Store-Pfad falsch -> klare Meldung ==")
try:
plan_store_renames(root / "gibtsnicht", "firma.de", "a", "b")
sys.exit(1)
except AbortedError as e:
print(f" ok: {str(e)[:70]}...")
print("\nALLE TESTS BESTANDEN")
tests/test_users_config.py
+149
View File
@@ -0,0 +1,149 @@
"""Testet UsersConfig gegen eine realistische users.cfg."""
import shutil, sys, tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from kerio_users_config import UsersConfig, validate_login
from kerio_common import AbortedError
SAMPLE = '''<config>
<list name="User">
<listitem>
<variable name="Name">Admin</variable>
<variable name="Domain">local.lan</variable>
<variable name="Account_enabled">1</variable>
<variable name="Guid">CDE8D0A3-929B-4AC6-8197-4C39D58C8A64</variable>
<variable name="Password">D3S:0faed6fd</variable>
<variable name="Fullname">Administrator</variable>
</listitem>
<listitem>
<variable name="Name">max.mustermann</variable>
<variable name="Domain">local.lan</variable>
<variable name="Account_enabled">1</variable>
<variable name="Guid">94480090-4C04-4FE2-9126-0DBDAEC85F5F</variable>
<variable name="Password">D3S:b6a11602</variable>
<variable name="Fullname">Max Mustermann</variable>
</listitem>
<listitem>
<variable name="Name">max.mustermann</variable>
<variable name="Domain">andere.lan</variable>
<variable name="Account_enabled">1</variable>
<variable name="Guid">AAAAAAAA-0000-0000-0000-000000000000</variable>
<variable name="Password">D3S:anders</variable>
<variable name="Fullname">Max in anderer Domain</variable>
</listitem>
<listitem>
<variable name="Name">ldap.user</variable>
<variable name="Domain">local.lan</variable>
<variable name="Account_enabled">1</variable>
<variable name="Guid">BBBBBBBB-0000-0000-0000-000000000000</variable>
<variable name="Fullname">LDAP Benutzer</variable>
</listitem>
</list>
<list name="Group">
<listitem>
<variable name="Name">testgruppe</variable>
<variable name="Domain">local.lan</variable>
<variable name="Guid">E78FEFAB-2C65-4D08-946B-76E3989EEF9D</variable>
</listitem>
</list>
<list name="Alias">
</list>
<list name="UserAdditionalData">
<listitem>
<variable name="Name">max.mustermann</variable>
<variable name="Domain">local.lan</variable>
<variable name="InternalDb">1</variable>
<variable name="PasswordHistory">SHA:28108e32</variable>
</listitem>
<listitem>
<variable name="Name">ldap.user</variable>
<variable name="Domain">local.lan</variable>
<variable name="InternalDb">0</variable>
</listitem>
</list>
</config>
'''
tmp = Path(tempfile.mkdtemp())
cfg_path = tmp / "users.cfg"
cfg_path.write_text(SAMPLE, encoding="utf-8")
ok = True
def check(label, cond):
global ok
print(f" {'ok ' if cond else 'FEHLER'} {label}")
if not cond: ok = False
print("== 1. Lesen ==")
c = UsersConfig(cfg_path)
check("Domains gefunden", set(c.domains()) == {"local.lan", "andere.lan"})
check("4 Benutzer gesamt", len(c.users()) == 4)
check("3 in local.lan", len(c.users("local.lan")) == 3)
u = c.find_user("max.mustermann", "local.lan")
check("Benutzer gefunden", u is not None and u["guid"].startswith("94480090"))
check("fullName gelesen", u["fullName"] == "Max Mustermann")
check("Domain-Trennung: andere.lan hat eigenen Eintrag",
c.find_user("max.mustermann", "andere.lan")["guid"].startswith("AAAA"))
print("\n== 2. Verzeichnisdienst erkennen ==")
check("max.mustermann ist intern", c.is_internal("max.mustermann", "local.lan"))
check("ldap.user ist NICHT intern", not c.is_internal("ldap.user", "local.lan"))
print("\n== 3. Login-Namen validieren ==")
for good in ("anna.schmidt", "a", "a-b_c.d", "user123"):
try:
validate_login(good); check(f"{good!r} erlaubt", True)
except AbortedError:
check(f"{good!r} erlaubt", False)
for bad in ("", ".anfang", "ende.", "mit leerzeichen", "boes<xml>", "a"*100, "a&b"):
try:
validate_login(bad); check(f"{bad!r} abgelehnt", False)
except AbortedError:
check(f"{bad!r} abgelehnt", True)
print("\n== 4. Umbenennen ==")
c2 = UsersConfig(cfg_path)
n = c2.rename_user("max.mustermann", "maxi.musterfrau", "local.lan", "Maxi Musterfrau")
check("2 Bloecke geaendert (User + UserAdditionalData)", n == 2)
c2.save()
c3 = UsersConfig(cfg_path)
check("neuer Name da", c3.find_user("maxi.musterfrau", "local.lan") is not None)
check("alter Name weg in local.lan", c3.find_user("max.mustermann", "local.lan") is None)
check("fullName geaendert", c3.find_user("maxi.musterfrau","local.lan")["fullName"] == "Maxi Musterfrau")
check("Guid unveraendert", c3.find_user("maxi.musterfrau","local.lan")["guid"].startswith("94480090"))
check("ANDERE DOMAIN unberuehrt",
c3.find_user("max.mustermann", "andere.lan") is not None)
check("Passworthistorie mitgewandert", c3.is_internal("maxi.musterfrau", "local.lan"))
txt = cfg_path.read_text(encoding="utf-8")
check("Passwort-Hash erhalten", "D3S:b6a11602" in txt)
check("PasswordHistory erhalten", "SHA:28108e32" in txt)
check("Admin unberuehrt", "<variable name=\"Name\">Admin</variable>" in txt)
check("Gruppe unberuehrt", "testgruppe" in txt)
check("kein max.mustermann mehr in UserAdditionalData/local.lan",
txt.count("max.mustermann") == 1) # nur noch andere.lan
print("\n== 5. other_occurrences ==")
c4 = UsersConfig(cfg_path)
hits = c4.other_occurrences("testgruppe", "local.lan")
check("findet Treffer in Group-Liste", any("Group" in h for h in hits))
check("keine Treffer fuer unbekannten Namen",
c4.other_occurrences("gibtsnicht", "local.lan") == [])
print("\n== 6. Abgelehnter Login-Name aendert nichts ==")
c5 = UsersConfig(cfg_path)
before = c5.text
try:
c5.rename_user("maxi.musterfrau", "boes<xml>", "local.lan")
check("haette ablehnen muessen", False)
except AbortedError:
check("ungueltiger Zielname abgelehnt", True)
check("Text unveraendert", c5.text == before)
print("\n" + ("ALLE TESTS BESTANDEN" if ok else "FEHLGESCHLAGEN"))
sys.exit(0 if ok else 1)