Move LDAP related code to separate file (bbe828b9) · Commits · Samorząd Uczniowski / Wybory

voters/forms.py

+2 −84

Changes for voters/forms.py: 2 added lines, 84 removed lines.
Original line number	Diff line number	Diff line
		import ssl

		import ldap3
		from django import forms
		from django.conf import settings
		from django.contrib import messages
		from django.forms import ValidationError
		from django_admin_action_forms import (
		@@ -11,88 +7,10 @@ from django_admin_action_forms import (
		TabularAdminActionInline,
		)

		from voters.ldap import query as ldap_query
		from voters.models import Group, Voter


		def get_ldap_attr(obj: dict, attr: str) -> str:
		if attr not in obj["attributes"]:
		raise ValueError(f"LDAP: {obj['raw_dn']} has no {attr} attribute")
		val = obj["attributes"][attr]
		if isinstance(val, list):
		if len(val) == 0:
		raise ValueError(f"LDAP: {obj['raw_dn']} has no {attr} attribute")
		if len(val) > 1:
		raise ValueError(
		f"LDAP: {obj['raw_dn']} has more than one {attr} attribute"
		)
		val = val[0]
		if isinstance(val, bytes):
		val = val.decode("utf-8")
		if not isinstance(val, str):
		raise ValueError(
		f"LDAP: {obj['raw_dn']} attribute {attr} could not be converted to string."
		)
		return val


		def render_group_template(template: str, group: str) -> str:
		return template % {"group": ldap3.utils.conv.escape_filter_chars(group)}


		def voter_sort_key(voter):
		return voter["name"]


		def LDAPquery(groups: list[str]) -> dict[str, list[dict[str, str]] \| None]:
		result = {}
		with ldap3.Connection(
		server=ldap3.Server(
		host=settings.LDAP_HOST,
		port=settings.LDAP_PORT,
		tls=ldap3.Tls(
		ca_certs_file=settings.LDAP_TLS_CA_CERTS_FILE,
		validate=ssl.CERT_REQUIRED,
		)
		if settings.LDAP_TLS
		else None,
		use_ssl=False,
		),
		user=settings.LDAP_BIND_DN,
		password=settings.LDAP_BIND_PASSWORD,
		auto_bind=ldap3.AUTO_BIND_TLS_BEFORE_BIND
		if settings.LDAP_TLS
		else ldap3.AUTO_BIND_NO_TLS,
		client_strategy=ldap3.SAFE_SYNC,
		raise_exceptions=True,
		) as conn:
		for group in set(groups):
		try:
		_, _, response, _ = conn.search(
		search_base=render_group_template(
		settings.LDAP_GROUP_SEARCH_BASE_TEMPLATE, group
		),
		search_filter=render_group_template(
		settings.LDAP_GROUP_FILTER_TEMPLATE, group
		),
		attributes=ldap3.ALL_ATTRIBUTES,
		time_limit=settings.LDAP_TIMEOUT,
		)
		except ldap3.core.exceptions.LDAPNoSuchObjectResult:
		result[group] = None
		else:
		result[group] = sorted(
		[
		{
		"name": get_ldap_attr(obj, settings.LDAP_NAME_FIELD),
		"email": get_ldap_attr(obj, settings.LDAP_EMAIL_FIELD),
		}
		for obj in response
		],
		key=voter_sort_key,
		)
		return result


		class GroupVotersImportInlineForm(InlineActionForm):
		name = forms.CharField(label="Imię i nazwisko")
		email = forms.EmailField(label="Email")
		@@ -131,7 +49,7 @@ class LDAPImportActionForm(AdminActionForm):
		selected_ids = request.POST.getlist("_selected_action")
		queryset = Group.objects.filter(pk__in=selected_ids)

		search_results = LDAPquery([group.name for group in queryset])
		search_results = ldap_query([group.name for group in queryset])

		initial = []
		for group in queryset:

voters/ldap.py

0 → 100644

+79 −0

Changes for voters/ldap.py: 79 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import ssl

		import ldap3
		from django.conf import settings


		def get_attr(obj: dict, attr: str) -> str:
		if attr not in obj["attributes"]:
		raise ValueError(f"LDAP: {obj['raw_dn']} has no {attr} attribute")
		val = obj["attributes"][attr]
		if isinstance(val, list):
		if len(val) == 0:
		raise ValueError(f"LDAP: {obj['raw_dn']} has no {attr} attribute")
		if len(val) > 1:
		raise ValueError(
		f"LDAP: {obj['raw_dn']} has more than one {attr} attribute"
		)
		val = val[0]
		if isinstance(val, bytes):
		val = val.decode("utf-8")
		if not isinstance(val, str):
		raise ValueError(
		f"LDAP: {obj['raw_dn']} attribute {attr} could not be converted to string."
		)
		return val


		def render_group_template(template: str, group: str) -> str:
		return template % {"group": ldap3.utils.conv.escape_filter_chars(group)}


		def query(groups: list[str]) -> dict[str, list[dict[str, str]] \| None]:
		result = {}
		with ldap3.Connection(
		server=ldap3.Server(
		host=settings.LDAP_HOST,
		port=settings.LDAP_PORT,
		tls=ldap3.Tls(
		ca_certs_file=settings.LDAP_TLS_CA_CERTS_FILE,
		validate=ssl.CERT_REQUIRED,
		)
		if settings.LDAP_TLS
		else None,
		use_ssl=False,
		),
		user=settings.LDAP_BIND_DN,
		password=settings.LDAP_BIND_PASSWORD,
		auto_bind=ldap3.AUTO_BIND_TLS_BEFORE_BIND
		if settings.LDAP_TLS
		else ldap3.AUTO_BIND_NO_TLS,
		client_strategy=ldap3.SAFE_SYNC,
		raise_exceptions=True,
		) as conn:
		for group in set(groups):
		try:
		_, _, response, _ = conn.search(
		search_base=render_group_template(
		settings.LDAP_GROUP_SEARCH_BASE_TEMPLATE, group
		),
		search_filter=render_group_template(
		settings.LDAP_GROUP_FILTER_TEMPLATE, group
		),
		attributes=ldap3.ALL_ATTRIBUTES,
		time_limit=settings.LDAP_TIMEOUT,
		)
		except ldap3.core.exceptions.LDAPNoSuchObjectResult:
		result[group] = None
		else:
		result[group] = sorted(
		[
		{
		"name": get_attr(obj, settings.LDAP_NAME_FIELD),
		"email": get_attr(obj, settings.LDAP_EMAIL_FIELD),
		}
		for obj in response
		],
		key=lambda obj: (obj["name"], obj["email"]),
		)
		return result