Merge pull request #92 from mgorny/no-django_ldap_auth

Replace django_auth_ldap with own backend

8 files changed, 133 insertions, 21 deletions

diff --git a/okupy/accounts/views.py b/okupy/accounts/views.py
index ab96d87..36980ee 100644
--- a/okupy/accounts/views.py
+++ b/okupy/accounts/views.py
@@ -139,7 +139,10 @@ def login(request):
             it was successful. If it retrieves None then it failed to login
             """
             try:
-                user = authenticate(username=username, password=password)
+                user = authenticate(
+                    request=request,
+                    username=username,
+                    password=password)
             except Exception as error:
                 logger.critical(error, extra=log_extra_data(request))
                 logger_mail.exception(error)
diff --git a/okupy/common/auth.py b/okupy/common/auth.py
index aa238fc..9dcf554 100644
--- a/okupy/common/auth.py
+++ b/okupy/common/auth.py
@@ -5,14 +5,53 @@ from django.contrib.auth.backends import ModelBackend
 from django.db import IntegrityError
 
 from okupy.accounts.models import LDAPUser
+from okupy.common.ldap_helpers import get_bound_ldapuser
 
 from OpenSSL.crypto import load_certificate, FILETYPE_PEM
 
+import ldap
 import paramiko
 
 import base64
 
 
+class LDAPAuthBackend(ModelBackend):
+    """
+    Authentication backend that authenticates against LDAP password.
+    If authentication succeeds, it sets up secondary password
+    for the session.
+    """
+
+    def authenticate(self, request, username, password):
+        # LDAP is case- and whitespace-insensitive
+        # we do normalization to avoid duplicate django db entries
+        # and help mockldap
+        username = username.lower().strip()
+
+        try:
+            bound_ldapuser = get_bound_ldapuser(
+                request=request,
+                username=username,
+                password=password)
+
+            with bound_ldapuser as u:
+                UserModel = get_user_model()
+                attr_dict = {
+                    UserModel.USERNAME_FIELD: u.username
+                }
+
+                user = UserModel(**attr_dict)
+                try:
+                    user.save()
+                except IntegrityError:
+                    user = UserModel.objects.get(**attr_dict)
+                return user
+        except ldap.INVALID_CREDENTIALS:
+            return None
+        except ldap.STRONG_AUTH_REQUIRED:
+            return None
+
+
 class SSLCertAuthBackend(ModelBackend):
     """
     Authentication backend taht uses client certificate information.
diff --git a/okupy/common/ldap_helpers.py b/okupy/common/ldap_helpers.py
index 43f3e3e..c8ac5dd 100644
--- a/okupy/common/ldap_helpers.py
+++ b/okupy/common/ldap_helpers.py
@@ -8,14 +8,18 @@ from okupy import OkupyError
 from okupy.accounts.models import LDAPUser
 from okupy.crypto.ciphers import cipher
 
+from django.conf import settings #debug
+from django.db import connections
 
-def get_bound_ldapuser(request, password=None):
+
+def get_bound_ldapuser(request, password=None, username=None):
     """
     Get LDAPUser with connection bound to the current user.
     Uses either provided password or the secondary password saved
     in session.
     """
-    username = request.user.username
+    if not username:
+        username = request.user.username
     if not password:
         try:
             password = b64encode(cipher.decrypt(
diff --git a/okupy/settings/__init__.py b/okupy/settings/__init__.py
index bdada0a..76f5ba1 100644
--- a/okupy/settings/__init__.py
+++ b/okupy/settings/__init__.py
@@ -26,7 +26,7 @@ MESSAGE_STORAGE = 'django.contrib.messages.storage.session.SessionStorage'
 
 # Custom authentication backend
 AUTHENTICATION_BACKENDS = (
-    'django_auth_ldap.backend.LDAPBackend',
+    'okupy.common.auth.LDAPAuthBackend',
     'okupy.common.auth.SSLCertAuthBackend',
     'okupy.common.auth.SSHKeyAuthBackend',
 )
@@ -140,10 +140,6 @@ LOGGING = {
             'handlers': ['console' if DEBUG else 'syslog'],
             'level': 'DEBUG' if DEBUG else 'INFO',
         },
-        'django_auth_ldap': {
-            'handlers': ['console' if DEBUG else 'syslog'],
-            'level': 'DEBUG' if DEBUG else 'INFO',
-        },
         'paramiko': {
             'handlers': ['console' if DEBUG else 'syslog'],
             'level': 'DEBUG' if DEBUG else 'INFO',
diff --git a/okupy/tests/settings.py b/okupy/tests/settings.py
index 97b2844..deb2f15 100644
--- a/okupy/tests/settings.py
+++ b/okupy/tests/settings.py
@@ -26,7 +26,7 @@ MESSAGE_STORAGE = 'django.contrib.messages.storage.session.SessionStorage'
 
 # Custom authentication backend
 AUTHENTICATION_BACKENDS = (
-    'django_auth_ldap.backend.LDAPBackend',
+    'okupy.common.auth.LDAPAuthBackend',
     'okupy.common.auth.SSLCertAuthBackend',
     'okupy.common.auth.SSHKeyAuthBackend',
 )
@@ -50,7 +50,6 @@ INSTALLED_APPS = (
     'django.contrib.sites',
     'django.contrib.messages',
     'django.contrib.staticfiles',
-    'django_auth_ldap',
     'django_otp',
     'discover_runner',
     'okupy.accounts',
@@ -260,10 +259,6 @@ LOGGING = {
             'handlers': ['console' if DEBUG else 'null'],
             'level': 'DEBUG' if DEBUG else 'INFO',
         },
-        'django_auth_ldap': {
-            'handlers': ['console' if DEBUG else 'null'],
-            'level': 'DEBUG' if DEBUG else 'INFO',
-        },
     }
 }
 
diff --git a/okupy/tests/unit/test_auth.py b/okupy/tests/unit/test_auth.py
index 586987b..de0f367 100644
--- a/okupy/tests/unit/test_auth.py
+++ b/okupy/tests/unit/test_auth.py
@@ -4,6 +4,7 @@ from mockldap import MockLdap
 
 from django.conf import settings
 from django.contrib.auth import authenticate
+from django.contrib.auth.models import User
 from django.test import TestCase
 
 from okupy.common.test_helpers import ldap_users, set_request
@@ -112,3 +113,75 @@ class AuthSSHUnitTests(TestCase):
             data=base64.b64decode(vars.TEST_SSH_KEY_FOR_NO_USER))
         u = authenticate(ssh_key=key)
         self.assertIs(u, None)
+
+
+class AuthLDAPUnitTests(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.mockldap = MockLdap(vars.DIRECTORY)
+
+    @classmethod
+    def tearDownClass(cls):
+        del cls.mockldap
+
+    def setUp(self):
+        self.mockldap.start()
+        self.ldapobj = self.mockldap[settings.AUTH_LDAP_SERVER_URI]
+        self.request = set_request(uri='/login')
+
+    def tearDown(self):
+        self.mockldap.stop()
+        del self.ldapobj
+
+    def test_successful_login(self):
+        user = authenticate(request=self.request,
+                            username='alice', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
+
+    def test_successful_login_lowercase(self):
+        user = authenticate(request=self.request,
+                            username='Alice', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
+
+    def test_successful_login_trailing_whitespace(self):
+        user = authenticate(request=self.request,
+                            username='alice ', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
+
+    def test_successful_login_leading_whitespace(self):
+        user = authenticate(request=self.request,
+                            username=' alice', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
+
+    def test_failed_login_wrong_password(self):
+        user = authenticate(request=self.request,
+                            username='alice', password='ldaptest1')
+        self.assertIs(user, None)
+
+    def test_failed_login_wrong_username(self):
+        user = authenticate(request=self.request,
+                            username='wrong', password='ldaptest')
+        self.assertIs(user, None)
+
+    def test_add_user_in_db(self):
+        authenticate(request=self.request,
+                     username='alice', password='ldaptest')
+        self.assertEqual(User.objects.count(), 1)
+
+    def test_successful_login_existing_user(self):
+        User.objects.create(username='alice')
+        user = authenticate(request=self.request,
+                            username='alice', password='ldaptest')
+        self.assertTrue(user is not None)
+        self.assertEqual(User.objects.count(), 1)
+
+    def test_successful_login_existing_user_insensitive(self):
+        User.objects.create(username='alice')
+        user = authenticate(request=self.request,
+                            username='Alice', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
+
+    def test_successful_login_insensitive(self):
+        user = authenticate(request=self.request,
+                            username='Alice', password='ldaptest')
+        self.assertTrue(user.username, 'alice')
diff --git a/okupy/tests/unit/test_login.py b/okupy/tests/unit/test_login.py
index f781741..c9948db 100644
--- a/okupy/tests/unit/test_login.py
+++ b/okupy/tests/unit/test_login.py
@@ -52,7 +52,7 @@ class LoginUnitTests(OkupyTestCase):
 
     @no_database()
     @override_settings(AUTHENTICATION_BACKENDS=(
-        'django_auth_ldap.backend.LDAPBackend',
+        'okupy.common.auth.LDAPAuthBackend',
         'django.contrib.auth.backends.ModelBackend'))
     def test_no_database_raises_critical(self):
         request = set_request(uri='/login', post=vars.LOGIN_ALICE,
@@ -64,7 +64,7 @@ class LoginUnitTests(OkupyTestCase):
 
     @no_database()
     @override_settings(AUTHENTICATION_BACKENDS=(
-        'django_auth_ldap.backend.LDAPBackend',
+        'okupy.common.auth.LDAPAuthBackend',
         'django.contrib.auth.backends.ModelBackend'))
     def test_no_database_sends_notification_mail(self):
         request = set_request(uri='/login', post=vars.LOGIN_ALICE,
@@ -144,17 +144,20 @@ class LoginUnitTestsNoLDAP(OkupyTestCase):
         self.assertMessage(response, 'Login failed', 40)
 
     def test_dont_authenticate_from_db_when_ldap_is_down(self):
-        request = set_request(uri='/login', post=vars.LOGIN_BOB, messages=True)
+        request = set_request(uri='/login', post=vars.LOGIN_BOB,
+                              messages=True)
         response = login(request)
         response.context = RequestContext(request)
-        self.assertMessage(response, 'Login failed', 40)
+        self.assertMessage(response,
+            "Can't contact the LDAP server or the database", 40)
 
-    def test_no_ldap_connection_raises_login_failed_in_login(self):
+    def test_no_ldap_connection_raises_ldaperror_in_login(self):
         request = set_request(uri='/login', post=vars.LOGIN_WRONG,
                               messages=True)
         response = login(request)
         response.context = RequestContext(request)
-        self.assertMessage(response, 'Login failed', 40)
+        self.assertMessage(response,
+            "Can't contact the LDAP server or the database", 40)
 
     def test_no_ldap_connection_in_logout_sends_notification_mail(self):
         request = set_request(uri='/login', post=vars.LOGIN_ALICE,
diff --git a/requirements/base.txt b/requirements/base.txt
index f63e9ab..8747082 100644
--- a/requirements/base.txt
+++ b/requirements/base.txt
@@ -1,5 +1,4 @@
 django>=1.5
-django-auth-ldap>=1.1.4
 django-compressor>=1.3
 django-otp>=0.1.7
 git+https://github.com/gentoo/django-ldapdb@okupy_v1#egg=django-ldapdb