alientim/tesm
0
0 Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 116 Wiki Activity

Verzeichnisanmeldung: Wiederholversuch, weniger Runden, Zeiten

Browse Source
This commit is contained in:
alientim committed 2026-09-12 23:46:51 +02:00
1 parent 86c09d8a0f
commit 2fd78ac26b
3 files changed
+109 -5
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/tesm-core/pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "tesm-core"
version = "2.8.1"
version = "2.8.2"
description = "Gemeinsames Anwendungsgeruest (Auth, RBAC, Navigation, Design-System, Migrationen) fuer TESM und TESM-Lizenzserver"
requires-python = ">=3.11"
dependencies = [
packages/tesm-core/src/tesm_core/auth/directory.py
+107 -3
View File
@@ -26,6 +26,7 @@ import re
import socket
import sqlite3
import ssl
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any, Literal
@@ -162,19 +163,82 @@ def _tls_object(config: dict[str, Any]):
def _server(config: dict[str, Any]):
"""Das Server-Objekt -- so sparsam wie moeglich.
``get_info`` entscheidet, was ldap3 beim Verbinden **zusaetzlich** holt.
Mit ``ALL`` ist das die Beschreibung des Verzeichnisses *und sein ganzes
Schema*: bei einem Active Directory sind das Tausende Attribut- und
Klassendefinitionen, mehrere Megabyte, bei jeder Verbindung neu. Fuer eine
Anmeldung braucht davon nichts -- gelesen wird hoechstens der
``defaultNamingContext``, wenn kein Basis-DN eingetragen ist, und dafuer
genuegt ``DSA`` (nur die Serverinformation, ein Datensatz).
Liegt der Domain-Controller in einem anderen Netz, ist genau das der
Unterschied zwischen "geht sofort" und "dauert manchmal".
"""
from ldap3 import Server
security = str(config["ldap_security"])
braucht_basis = not str(config.get("ldap_base_dn") or "").strip()
return Server(
str(config["ldap_host"]),
port=int(config["ldap_port"]),
use_ssl=security == "ldaps",
tls=_tls_object(config) if security in {"ldaps", "starttls"} else None,
get_info="ALL",
get_info="DSA" if braucht_basis else "NO_INFO",
connect_timeout=int(config["ldap_timeout"]),
)
#: Fehler, die nur "gerade jetzt nicht" bedeuten -- nicht "falsches Passwort".
#:
#: Die Liste stammt aus der Fassung 1.2.8, wo dieses Problem schon einmal
#: geloest war. Der Kommentar dort beschreibt es genauer, als ich es koennte:
#: ein Domain-Controller in einem anderen, langsameren Netzsegment -- die
#: allererste Verbindung dorthin ist oft spuerbar langsamer als jede folgende,
#: und der erste Login schlug mit "Benutzer/Passwort ungueltig" fehl, waehrend
#: ein sofortiger zweiter Versuch klappte.
TRANSIENTE_FEHLER: tuple[type[Exception], ...] = ()
def _transiente_fehler() -> tuple[type[Exception], ...]:
"""Die Ausnahmeklassen von ldap3, die einen Wiederholversuch verdienen."""
global TRANSIENTE_FEHLER
if TRANSIENTE_FEHLER:
return TRANSIENTE_FEHLER
from ldap3.core.exceptions import (
LDAPResponseTimeoutError,
LDAPSocketOpenError,
LDAPSocketReceiveError,
LDAPSocketSendError,
)
TRANSIENTE_FEHLER = (
LDAPSocketOpenError,
LDAPSocketReceiveError,
LDAPSocketSendError,
LDAPResponseTimeoutError,
)
return TRANSIENTE_FEHLER
def mit_wiederholung(aufbau: Any) -> Any:
"""Genau ein zweiter Versuch -- und nur bei einem Netzfehler.
Nicht bei einer abgelehnten Anmeldung: ein Tippfehler im Passwort wuerde
sonst den Sperrzaehler des Kontos zweimal statt einmal hochzaehlen. Das
ist die Regel aus 1.2.8, und sie ist richtig.
"""
try:
return aufbau()
except _transiente_fehler() as fehler:
logger.info(
"Verzeichnisdienst antwortete nicht (%s) -- ein zweiter Versuch",
type(fehler).__name__,
)
return aufbau()
def _connect(config: dict[str, Any], user: str, password: str):
"""Baut eine Verbindung auf und bindet. Wirft bei Netz-/TLS-Problemen."""
from ldap3 import Connection
@@ -192,6 +256,12 @@ def _connect(config: dict[str, Any], user: str, password: str):
auto_bind=False,
raise_exceptions=False,
receive_timeout=int(config["ldap_timeout"]),
# Ein Active Directory antwortet auf eine Suche ab der Domaenenwurzel
# mit Verweisen auf andere Controller. ldap3 folgt ihnen von sich aus
# -- und baut dafuer neue Verbindungen auf, womoeglich in ein anderes
# Netz. Fuer eine Anmeldung ist das nicht nur unnoetig, sondern die
# Quelle genau der Wartezeiten, die niemand erklaeren kann.
auto_referrals=False,
)
if security == "starttls":
if not connection.open():
@@ -253,8 +323,19 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
if not config["ldap_verify_cert"]:
warnings.append("Serverzertifikat des Verzeichnisdienstes wird nicht geprüft")
begonnen = time.monotonic()
zeiten: list[str] = []
def gemessen(name: str, seit: float) -> float:
"""Haelt eine Stufe mit ihrer Dauer fest und gibt die neue Marke."""
jetzt = time.monotonic()
zeiten.append(f"{name} {int((jetzt - seit) * 1000)}ms")
return jetzt
try:
service_conn, bound = _connect(config, bind_user, bind_password)
service_conn, bound = mit_wiederholung(
lambda: _connect(config, bind_user, bind_password)
)
except (PermissionError, ConnectionError) as exc:
logger.error("Verzeichnisdienst: %s", exc)
return DirectoryResult("unavailable", message=str(exc), warnings=tuple(warnings))
@@ -271,8 +352,10 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
warnings=tuple(warnings),
)
marke = gemessen("Dienstbindung", begonnen)
try:
base = _base_dn(service_conn, config)
marke = gemessen("Basis-DN", marke)
if not base:
return DirectoryResult(
"unavailable", message="Basis-DN konnte nicht ermittelt werden.", warnings=tuple(warnings)
@@ -288,8 +371,10 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
attributes=list(dict.fromkeys(wanted)),
size_limit=2,
)
marke = gemessen("Benutzersuche", marke)
entries = list(service_conn.entries)
if len(entries) != 1:
logger.info("Verzeichnisanmeldung %s: %s", username, ", ".join(zeiten))
return DirectoryResult("invalid", warnings=tuple(warnings))
entry = entries[0]
@@ -300,9 +385,18 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
# Zweite, unabhaengige Verbindung mit den Benutzerdaten -- die eigentliche
# Passwortpruefung. Niemals ueber die Dienstkonto-Verbindung rebinden.
user_conn, user_bound = _connect(config, dn, password)
#
# Mit Wiederholversuch, und genau hier zeigte sich das Symptom: bei
# einem Domain-Controller in einem anderen Netz schlug der erste
# Versuch mit "Benutzer/Passwort ungueltig" fehl, der zweite gleich
# darauf klappte. Wiederholt wird nur bei einem Netzfehler -- eine
# abgelehnte Anmeldung bleibt abgelehnt, sonst zaehlt ein Tippfehler
# den Sperrzaehler des Kontos zweimal hoch.
user_conn, user_bound = mit_wiederholung(lambda: _connect(config, dn, password))
marke = gemessen("Passwortpruefung", marke)
try:
if not user_bound:
logger.info("Verzeichnisanmeldung %s: %s", username, ", ".join(zeiten))
return DirectoryResult("invalid", warnings=tuple(warnings))
finally:
try:
@@ -313,6 +407,7 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
group_dns: tuple[str, ...] = ()
if config["ldap_group_sync"]:
group_dns = _resolve_groups(service_conn, base, dn)
marke = gemessen("Gruppen", marke)
profile = DirectoryProfile(
username=_attr(entry, attribute) or username,
@@ -322,6 +417,15 @@ def authenticate(conn: sqlite3.Connection, username: str, password: str) -> Dire
email=_attr(entry, email_attribute),
group_dns=group_dns,
)
# Eine Zeile je Anmeldung mit den Dauern der Stufen. Aus "dauert
# manchmal laenger" wird damit eine Zahl -- und man sieht, ob es die
# Verbindung ist, die Suche oder die Gruppen.
logger.info(
"Verzeichnisanmeldung %s: %s, gesamt %dms",
username,
", ".join(zeiten),
int((time.monotonic() - begonnen) * 1000),
)
return DirectoryResult("ok", profile=profile, warnings=tuple(warnings))
finally:
try: