"""LDAP/AD authentication for dashboard users. User-bind design (no service account required): 1. The user binds with their own credentials as ``user@domain`` (AD accepts userPrincipalName as bind DN). 2. Over that authenticated connection we search the user's own directory entry, restricted by the in-chain matching rule to members of the configured admin group (direct or nested membership). Only members of the configured group may log in. All other users are rejected before any local session is created. """ import ssl from dataclasses import dataclass import ldap3 from ldap3.core.exceptions import LDAPExceptionError from ldap3.utils.conv import escape_filter_chars from app.core.config import Settings # LDAP_MATCHING_RULE_IN_CHAIN: resolves nested group membership server-side _MATCHING_RULE_IN_CHAIN = "1.2.840.113556.1941" class LdapAuthError(Exception): """Authentication failed. Message is safe to show to the user.""" def __init__(self, detail: str = "Benutzername oder Passwort falsch") -> None: super().__init__(detail) self.detail = detail @dataclass class LdapUser: username: str full_name: str | None email: str | None is_admin: bool # always True today: only admin-group members may log in def _build_server(settings: Settings) -> ldap3.Server: use_ssl = settings.ldap_uri.lower().startswith("ldaps://") # DC certificates come from an internal AD CA that is not distributed to # clients; full chain validation would need the enterprise root CA. tls = ldap3.Tls(validate=ssl.CERT_NONE) if use_ssl else None return ldap3.Server( settings.ldap_uri, get_info=ldap3.NONE, connect_timeout=settings.ldap_timeout_seconds, tls=tls, ) def _bind_dn(username: str, settings: Settings) -> str: if settings.ldap_domain: return f"{username}@{settings.ldap_domain}" return username def authenticate(username: str, password: str, settings: Settings) -> LdapUser: """Validate credentials against AD and verify admin-group membership. Raises LdapAuthError on any failure (wrong password, unknown user, not in admin group, LDAP unreachable). """ if not settings.ldap_enabled: raise LdapAuthError("LDAP ist nicht aktiviert") if not settings.ldap_admin_group: raise LdapAuthError("LDAP-Admin-Gruppe ist nicht konfiguriert") if not username or not password: raise LdapAuthError() server = _build_server(settings) try: conn = ldap3.Connection( server, user=_bind_dn(username.strip(), settings), password=password, auto_bind=True, receive_timeout=settings.ldap_timeout_seconds, ) except LDAPExceptionError as exc: # Covers invalidCredentials (49) as well as connect/TLS timeouts raise LdapAuthError() from exc with conn: user_filter = settings.ldap_user_filter.format( username=escape_filter_chars(username.strip()) ) # One search proves: user exists AND is (transitively) in the admin # group. Non-members simply match nothing. group_filter = f"(memberOf:{_MATCHING_RULE_IN_CHAIN}:={settings.ldap_admin_group})" ok = conn.search( search_base=settings.ldap_user_search_base, search_filter=f"(&{user_filter}{group_filter})", search_scope=ldap3.SUBTREE, attributes=["displayName", "mail"], ) if not ok or not conn.entries: raise LdapAuthError("Kein Zugriff: Kein Mitglied der Administrator-Gruppe") entry = conn.entries[0] return LdapUser( username=username.strip(), full_name=str(entry.displayName) if entry.displayName else None, email=str(entry.mail) if entry.mail else None, is_admin=True, )