mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-11-10 08:33:53 +01:00
LDAP synchronization process (#1732)
* Initial commit for LDAP sync process. * Fixed unit tests. * Added doc and run ldapsync tests on travis. * Removed server address. * Force bytes when calling ldap methods. * Fixed tests compat with python2. * Do not force bytes here. * Force bytes. * Force bytes for attribute names too. * Disable bytes mode in PY2. * Handle account removals. * Fixed unit tests. * Limit sync to simple users. * Fixed unit test. * Fixed unit test.
This commit is contained in:
Antoine Nguyen
@@ -0,0 +1 @@
|
||||
default_app_config = "modoboa.ldapsync.apps.LdapsyncConfig"
|
||||
@@ -0,0 +1,8 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class LdapsyncConfig(AppConfig):
|
||||
name = 'modoboa.ldapsync'
|
||||
|
||||
def ready(self):
|
||||
from . import handlers # noqa
|
||||
@@ -0,0 +1,36 @@
|
||||
"""LDAP sync related handlers."""
|
||||
|
||||
from django.db.models import signals
|
||||
from django.dispatch import receiver
|
||||
|
||||
from modoboa.core import models as core_models
|
||||
from modoboa.parameters import tools as param_tools
|
||||
|
||||
from . import lib
|
||||
|
||||
|
||||
@receiver(signals.post_save, sender=core_models.User)
|
||||
def sync_ldap_account(sender, instance, created, **kwargs):
|
||||
"""Create/modify a new LDAP account if needed."""
|
||||
config = dict(param_tools.get_global_parameters("core"))
|
||||
if not config["ldap_enable_sync"]:
|
||||
return
|
||||
if created:
|
||||
return
|
||||
if instance.role != "SimpleUsers":
|
||||
return
|
||||
update_fields = kwargs.get("update_fields")
|
||||
if update_fields and "last_login" in update_fields:
|
||||
return
|
||||
lib.update_ldap_account(instance, config)
|
||||
|
||||
|
||||
@receiver(signals.pre_delete, sender=core_models.User)
|
||||
def delete_ldap_account(sender, instance, **kwargs):
|
||||
"""Delete LDAP account if needed."""
|
||||
config = dict(param_tools.get_global_parameters("core"))
|
||||
if not config["ldap_enable_sync"]:
|
||||
return
|
||||
if instance.role != "SimpleUsers":
|
||||
return
|
||||
lib.delete_ldap_account(instance, config)
|
||||
@@ -0,0 +1,127 @@
|
||||
"""LDAP related functions."""
|
||||
|
||||
from __future__ import unicode_literals
|
||||
|
||||
import ldap
|
||||
import ldap.modlist as modlist
|
||||
|
||||
from django.utils.encoding import force_bytes, force_str
|
||||
from django.utils.translation import ugettext as _
|
||||
|
||||
from modoboa.lib.exceptions import InternalError
|
||||
|
||||
|
||||
def get_connection(config, username=None, password=None):
|
||||
"""Get a new connection to the LDAP directory."""
|
||||
uri = "{}:{}".format(
|
||||
config["ldap_server_address"], config["ldap_server_port"])
|
||||
uri = "{}://{}".format(
|
||||
"ldaps" if config["ldap_secured"] == "ssl" else "ldap", uri)
|
||||
conn = ldap.initialize(uri)
|
||||
conn.set_option(ldap.OPT_X_TLS_DEMAND, True)
|
||||
conn.set_option(ldap.OPT_DEBUG_LEVEL, 255)
|
||||
conn.simple_bind_s(
|
||||
force_str(username if username else config["ldap_sync_bind_dn"]),
|
||||
force_str(password if password else config["ldap_sync_bind_password"])
|
||||
)
|
||||
return conn
|
||||
|
||||
|
||||
def get_user_password(user, disable=False):
|
||||
"""Return ready-to-use password from user instance."""
|
||||
scheme, password = user.password.split("}")
|
||||
return (
|
||||
force_bytes(scheme) +
|
||||
b"}" +
|
||||
b"#" if disable else b"" +
|
||||
force_bytes(password)
|
||||
)
|
||||
|
||||
|
||||
def create_ldap_account(user, dn, conn):
|
||||
"""Create new account."""
|
||||
scheme, password = user.password.split("}")
|
||||
attrs = {
|
||||
"objectClass": [
|
||||
force_bytes("inetOrgPerson"), force_bytes("organizationalPerson")
|
||||
],
|
||||
"uid": [force_bytes(user.username)],
|
||||
"sn": [force_bytes(user.last_name)],
|
||||
"givenName": [force_bytes(user.first_name)],
|
||||
"cn": [force_bytes(user.username)],
|
||||
"displayName": [force_bytes(user.fullname)],
|
||||
"mail": [
|
||||
force_bytes(user.email), force_bytes(user.secondary_email)],
|
||||
"homePhone": [force_bytes(user.phone_number)],
|
||||
"userPassword": [get_user_password(user)]
|
||||
}
|
||||
ldif = modlist.addModlist(attrs)
|
||||
try:
|
||||
conn.add_s(dn, ldif)
|
||||
except ldap.LDAPError as e:
|
||||
raise InternalError(
|
||||
_("Failed to create LDAP account: {}").format(e)
|
||||
)
|
||||
|
||||
|
||||
def check_if_dn_exists(conn, dn):
|
||||
"""Check if DN already exists in directory."""
|
||||
try:
|
||||
res = conn.search_s(
|
||||
force_str(dn), ldap.SCOPE_SUBTREE,
|
||||
force_str("(&(objectClass=inetOrgPerson))")
|
||||
)
|
||||
res = res[0][0]
|
||||
except ldap.LDAPError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def update_ldap_account(user, config):
|
||||
"""Update existing account."""
|
||||
dn = config["ldap_sync_account_dn_template"] % {"user": user.username}
|
||||
conn = get_connection(config)
|
||||
if not check_if_dn_exists(conn, dn):
|
||||
create_ldap_account(user, dn, conn)
|
||||
return
|
||||
password = get_user_password(user, not user.is_active)
|
||||
ldif = [
|
||||
(ldap.MOD_REPLACE, "uid", force_bytes(user.username)),
|
||||
(ldap.MOD_REPLACE, "sn", force_bytes(user.last_name)),
|
||||
(ldap.MOD_REPLACE, "givenName", force_bytes(user.first_name)),
|
||||
(ldap.MOD_REPLACE, "cn", force_bytes(user.username)),
|
||||
(ldap.MOD_REPLACE, "displayName", force_bytes(user.fullname)),
|
||||
(ldap.MOD_REPLACE, "mail", force_bytes(user.email)),
|
||||
(ldap.MOD_REPLACE, "homePhone", force_bytes(user.phone_number)),
|
||||
(ldap.MOD_REPLACE, "userPassword", password)
|
||||
]
|
||||
try:
|
||||
conn.modify_s(dn, ldif)
|
||||
except ldap.LDAPError as e:
|
||||
raise InternalError(
|
||||
_("Failed to update LDAP account: {}").format(e))
|
||||
|
||||
|
||||
def delete_ldap_account(user, config):
|
||||
"""Delete remote LDAP account."""
|
||||
dn = config["ldap_sync_account_dn_template"] % {"user": user.username}
|
||||
conn = get_connection(config)
|
||||
if not check_if_dn_exists(conn, dn):
|
||||
return
|
||||
if config["ldap_sync_delete_remote_account"]:
|
||||
try:
|
||||
conn.delete_s(dn)
|
||||
except ldap.LDAPError as e:
|
||||
raise InternalError(
|
||||
_("Failed to delete LDAP account: {}").format(e)
|
||||
)
|
||||
else:
|
||||
password = get_user_password(user, True)
|
||||
ldif = [
|
||||
(ldap.MOD_REPLACE, "userPassword", password)
|
||||
]
|
||||
try:
|
||||
conn.modify_s(dn, ldif)
|
||||
except ldap.LDAPError as e:
|
||||
raise InternalError(
|
||||
_("Failed to disable LDAP account: {}").format(e))
|
||||
@@ -0,0 +1,105 @@
|
||||
"""LDAP sync. related tests."""
|
||||
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from unittest import skipIf
|
||||
|
||||
import ldap
|
||||
|
||||
from django.utils import six
|
||||
from django.utils.encoding import force_bytes, force_str
|
||||
|
||||
from modoboa.core import factories as core_factories
|
||||
from modoboa.lib.tests import NO_LDAP, ModoTestCase
|
||||
from modoboa.parameters import tools as param_tools
|
||||
|
||||
from . import lib
|
||||
|
||||
|
||||
@skipIf(NO_LDAP, "No ldap module installed")
|
||||
class LDAPSyncTestCase(ModoTestCase):
|
||||
"""Synchronization related tests."""
|
||||
|
||||
def setUp(self):
|
||||
super(LDAPSyncTestCase, self).setUp()
|
||||
self.set_global_parameters({
|
||||
"ldap_enable_sync": True,
|
||||
"ldap_server_port": 3389,
|
||||
"ldap_sync_bind_dn": "cn=admin,dc=example,dc=com",
|
||||
"ldap_sync_bind_password": "test",
|
||||
"ldap_sync_account_dn_template": (
|
||||
"cn=%(user)s,ou=users,dc=example,dc=com"),
|
||||
}, app="core")
|
||||
self.config = dict(param_tools.get_global_parameters("core"))
|
||||
self.conn = lib.get_connection(self.config)
|
||||
self.username = "testldap@test.com"
|
||||
self.dn = self.config["ldap_sync_account_dn_template"] % {
|
||||
"user": self.username}
|
||||
|
||||
def reset_ldap_directory(self):
|
||||
try:
|
||||
tmp_dn = force_bytes(self.dn) if six.PY2 else self.dn
|
||||
self.conn.delete_s(tmp_dn)
|
||||
except ldap.NO_SUCH_OBJECT:
|
||||
pass
|
||||
|
||||
def test_sync_user(self):
|
||||
self.reset_ldap_directory()
|
||||
user = core_factories.UserFactory(
|
||||
username=self.username,
|
||||
first_name="Test",
|
||||
last_name="LDAP",
|
||||
groups=("SimpleUsers",)
|
||||
)
|
||||
self.assertTrue(lib.check_if_dn_exists(self.conn, self.dn))
|
||||
|
||||
lib.get_connection(self.config, self.dn, "toto")
|
||||
|
||||
user.last_name = "LDAP Modif"
|
||||
user.save()
|
||||
lib.get_connection(self.config, self.dn, "toto")
|
||||
|
||||
res = self.conn.search_s(
|
||||
force_str(self.dn), ldap.SCOPE_SUBTREE,
|
||||
force_str("(&(objectClass=inetOrgPerson))")
|
||||
)
|
||||
self.assertIn(force_bytes(user.last_name), res[0][1]["sn"])
|
||||
|
||||
def test_sync_domainadmin(self):
|
||||
self.reset_ldap_directory()
|
||||
core_factories.UserFactory(
|
||||
username=self.username,
|
||||
first_name="Test",
|
||||
last_name="LDAP",
|
||||
groups=("DomainAdmins", )
|
||||
)
|
||||
self.assertFalse(lib.check_if_dn_exists(self.conn, self.dn))
|
||||
|
||||
def test_delete_user(self):
|
||||
self.reset_ldap_directory()
|
||||
user = core_factories.UserFactory(
|
||||
username=self.username,
|
||||
first_name="Test",
|
||||
last_name="LDAP",
|
||||
groups=("SimpleUsers",)
|
||||
)
|
||||
user.delete()
|
||||
ldap_record = self.conn.search_s(
|
||||
force_str(self.dn), ldap.SCOPE_SUBTREE,
|
||||
force_str("(&(objectClass=inetOrgPerson))")
|
||||
)
|
||||
password = ldap_record[0][1]["userPassword"][0].split(b"}")[1]
|
||||
self.assertTrue(password.startswith(b"#"))
|
||||
with self.assertRaises(ldap.INVALID_CREDENTIALS):
|
||||
lib.get_connection(self.config, self.dn, "toto")
|
||||
|
||||
user = core_factories.UserFactory(
|
||||
username=self.username,
|
||||
first_name="Test",
|
||||
last_name="LDAP",
|
||||
groups=("SimpleUsers",)
|
||||
)
|
||||
self.set_global_parameter(
|
||||
"ldap_sync_delete_remote_account", True, app="core")
|
||||
user.delete()
|
||||
self.assertFalse(lib.check_if_dn_exists(self.conn, self.dn))
|
||||
Reference in New Issue
Block a user