forked from ungleich-public/ungleich-user
Sajeyks
ab45535f38
python-ldap is a C extension that needs the OpenLDAP headers to build (and needed a workaround in the Dockerfile). ldap3, which the app already used for all writes, is pure Python. dal.backends.LdapBackend keeps the django-auth-ldap behaviour: search the uid in every LDAPSEARCH base as the admin DN, require exactly one match, log in with the user's DN, and mirror givenName, sn and mail into the local Django user. Empty passwords are rejected (LDAP treats them as anonymous logins, which succeed). LDAP_USE_TLS now applies StartTLS to every connection, including the admin connection used for writes, not just logins.
73 lines
2.7 KiB
Python
73 lines
2.7 KiB
Python
"""
|
|
LDAP authentication with ldap3, replacing django-auth-ldap and python-ldap.
|
|
|
|
Same behaviour as the old django-auth-ldap setup: the uid is searched (as
|
|
the admin DN) in every base of LDAPSEARCH, exactly one match is required,
|
|
the user logs in with their own DN, and a local Django user mirrors the
|
|
names and email from LDAP (AUTH_LDAP_USER_ATTR_MAP).
|
|
"""
|
|
import logging
|
|
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.contrib.auth.backends import ModelBackend
|
|
from ldap3.core.exceptions import LDAPException
|
|
from ldap3.utils.conv import escape_filter_chars
|
|
|
|
from .ungleich_ldap import LdapManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class LdapBackend(ModelBackend):
|
|
|
|
def authenticate(self, request, username=None, password=None, **kwargs):
|
|
# An empty password is an anonymous bind in LDAP and would succeed.
|
|
if not username or not password:
|
|
return None
|
|
|
|
ldap_manager = LdapManager()
|
|
try:
|
|
entry = self._find_user(ldap_manager, username)
|
|
if entry is None:
|
|
return None
|
|
conn = ldap_manager.get_conn(user=entry.entry_dn, password=password)
|
|
try:
|
|
if not conn.bind():
|
|
return None
|
|
finally:
|
|
conn.unbind()
|
|
except LDAPException as exc:
|
|
logger.error("LDAP authentication for %s failed: %s", username, exc)
|
|
return None
|
|
|
|
user = self._sync_user(username, entry)
|
|
return user if self.user_can_authenticate(user) else None
|
|
|
|
def _find_user(self, ldap_manager, username):
|
|
attributes = list(settings.AUTH_LDAP_USER_ATTR_MAP.values())
|
|
search_filter = "(uid={})".format(escape_filter_chars(username))
|
|
matches = []
|
|
conn = ldap_manager.get_admin_conn()
|
|
try:
|
|
for base in settings.LDAP_SEARCH_BASES:
|
|
conn.search(base, search_filter, attributes=attributes)
|
|
matches.extend(conn.entries)
|
|
finally:
|
|
conn.unbind()
|
|
if len(matches) != 1:
|
|
if matches:
|
|
logger.warning("uid %s matched %d entries; refusing login",
|
|
username, len(matches))
|
|
return None
|
|
return matches[0]
|
|
|
|
def _sync_user(self, username, entry):
|
|
user, created = get_user_model().objects.get_or_create(username=username)
|
|
for field, attr in settings.AUTH_LDAP_USER_ATTR_MAP.items():
|
|
value = entry[attr].value if attr in entry else None
|
|
setattr(user, field, value if isinstance(value, str) else "")
|
|
if created:
|
|
user.set_unusable_password()
|
|
user.save()
|
|
return user
|