duffyduckandClaude Opus 5
b12041fc81
kerio_trial_setup.py bringt eine frische Installation am Assistenten vorbei, wenn dessen Lizenzschritt nicht durchlaufbar ist: der Link "Im unregistrierten Modus aktivieren" fehlt je nach Version und die verlinkte Registrierungsseite ist seit der GFI-Uebernahme teils tot. Gesetzt wird ConfigWizardDone in mailserver.cfg; Kerio laeuft danach 30 Tage unregistriert, ohne Einschraenkung bei Benutzerverwaltung und Admin-API. Die Datei darf nur bei gestopptem Dienst bearbeitet werden, weil Kerio seine Konfiguration beim Beenden aus dem Speicher zurueckschreibt. Das Script stoppt den Dienst, liest danach neu ein, schreibt atomar ueber eine temporaere Datei und liest den Wert nach dem Start zur Kontrolle erneut. Backup vor jeder Aenderung, --restore spielt es zurueck. API-Client und Dienststeuerung wandern nach kerio_common.py, damit beide Scripts sie teilen. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
302 lines
10 KiB
Python
302 lines
10 KiB
Python
"""
|
|
Gemeinsame Bausteine fuer die Kerio-Connect-Scripts:
|
|
|
|
* KerioAdminApi - JSON-RPC-Client fuer die Administration API
|
|
* ServiceController - startet und stoppt den Kerio-Dienst
|
|
* Hilfsfunktionen zum Auffinden von Domains und Benutzern
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import http.cookiejar
|
|
import json
|
|
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_INSTALL_DIR = "/opt/kerio/mailserver"
|
|
DEFAULT_STORE_DIR = "/opt/kerio/mailserver/store"
|
|
DEFAULT_SERVICE = "kerio-connect"
|
|
|
|
|
|
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."""
|
|
|
|
|
|
# ==========================================================================
|
|
# 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
|
|
|
|
|
|
# ==========================================================================
|
|
# Dienststeuerung
|
|
# ==========================================================================
|
|
|
|
|
|
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 der Aenderung steht.
|
|
"""
|
|
|
|
def __init__(self, name: str = DEFAULT_SERVICE, mode: str = "auto",
|
|
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":
|
|
return self._run(f"/etc/init.d/{self.name}", "status").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":
|
|
if self._run("systemctl", "cat", self.name).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: "
|
|
f"{result.stderr.strip() or result.stdout.strip()}"
|
|
)
|
|
|
|
# Warten, bis der Dienst wirklich unten ist - sonst schreibt Kerio
|
|
# noch in Dateien, die wir gerade anfassen.
|
|
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.")
|