Search for LDAP entry by both username and e-mail + large refactor

README.md

@@ -1,64 +1,77 @@
-Taiga contrib ldap auth
-=======================
+# Taiga contrib ldap auth
 
-The Taiga plugin for ldap authentication.
+Taiga.io plugin for LDAP authentication.
 
-Installation
-------------
 
-### Taiga Back
+## Installation
 
-In your Taiga back python virtualenv install the pip package
-`taiga-contrib-ldap-auth` with:
+Install the PIP package `taiga-contrib-ldap-auth` in your
+`taiga-back` python virtualenv:
 
 ```bash
   pip install taiga-contrib-ldap-auth
 ```
 
-Modify your settings/local.py and include it on `INSTALLED_APPS` and add your
-LDAP configuration:
+If needed, change `pip` to `pip3` to use the Python 3 version.
+
+
+## Configuration
+
+### taiga-back
+
+Add the following to `settings/local.py`:
 
 ```python
-  INSTALLED_APPS += ["taiga_contrib_ldap_auth"]
-
-  LDAP_SERVER = 'ldap://ldap.example.com'
-  LDAP_PORT = 389
-
-  # Full DN of the service account use to connect to LDAP server and search for login user's account entry
-  # If LDAP_BIND_DN is not specified, or is blank, then an anonymous bind is attempated
-  LDAP_BIND_DN = 'CN=SVC Account,OU=Service Accounts,OU=Servers,DC=example,DC=com'
-  LDAP_BIND_PASSWORD = 'replace_me'   # eg.
-  # Starting point within LDAP structure to search for login user
-  LDAP_SEARCH_BASE = 'OU=DevTeam,DC=example,DC=net'
-  # LDAP property used for searching, ie. login username needs to match value in sAMAccountName property in LDAP
-  LDAP_SEARCH_PROPERTY = 'sAMAccountName'
-  LDAP_SEARCH_SUFFIX = None # '@example.com'
-
-  # Names of LDAP properties on user account to get email and full name
-  LDAP_EMAIL_PROPERTY = 'mail'
-  LDAP_FULL_NAME_PROPERTY = 'name'
-```
+INSTALLED_APPS += ["taiga_contrib_ldap_auth"]
 
-The logic of the code is such that a dedicated domain service account user performs a search on LDAP for an account that has a LDAP_SEARCH_PROPERTY value that matches the username the user typed in on the Taiga login form.  
-If the search is successful, then the code uses this value and the typed-in password to attempt a bind to LDAP using these credentials.
-If the bind is successful, then we can say that the user is authorised to log in to Taiga.
+LDAP_SERVER = 'ldap://ldap.example.com'
+LDAP_PORT = 389
 
-Optionally LDAP_SEARCH_SUFFIX can be set to allow for the search to match only the beginning of a field containing e.g. an email address.
+# Full DN of the service account use to connect to LDAP server and search for login user's account entry
+# If LDAP_BIND_DN is not specified, or is blank, then an anonymous bind is attempated
+LDAP_BIND_DN = 'CN=SVC Account,OU=Service Accounts,OU=Servers,DC=example,DC=com'
+LDAP_BIND_PASSWORD = '<REPLACE_ME>'
+
+# Starting point within LDAP structure to search for login user
+LDAP_SEARCH_BASE = 'OU=DevTeam,DC=example,DC=net'
+
+# Additional search criteria to the filter (will be ANDed)
+#LDAP_SEARCH_FILTER_ADDITIONAL = '(mail=*)'
+
+# Names of attributes to get username, e-mail and full name values from
+LDAP_USERNAME_ATTRIBUTE = 'uid'
+LDAP_EMAIL_ATTRIBUTE = 'mail'
+LDAP_FULL_NAME_ATTRIBUTE = 'displayName'
+```
 
-If the LDAP_BIND_DN configuration setting is not specified or is blank, then an anonymous bind is attempted to search for the login user's LDAP account entry.
+A dedicated domain service account user (specified by `LDAP_BIND_DN`)
+performs a search on LDAP for an account that has a
+`LDAP_USERNAME_ATTRIBUTE` or `LDAP_EMAIL_ATTRIBUTE` matching the
+user-provided login.
 
+If the search is successful, then the returned entry and the
+user-provided password  are used to attempt a bind to LDAP. If the bind is
+successful, then we can say that the user is authorised to log in to
+Taiga.
 
-RECOMMENDATION: Note that if you are using a service account for performing the LDAP search for the user that is logging on to Taiga, for security reasons, the service account user should be configured to only allow reading/searching the LDAP structure. No other LDAP (or wider network) permissions should be granted for this user because you need to specify the service account password in this file.
-A suitably strong password should be chosen, eg. VmLYBbvJaf2kAqcrt5HjHdG6
+If the `LDAP_BIND_DN` configuration setting is not specified or is
+blank, then an anonymous bind is attempted to search for the login
+user's LDAP account entry.
 
+**RECOMMENDATION**: for security reasons, if you are using a service
+account for performing the LDAP search, it should be configured to only
+allow reading/searching the LDAP structure. No other LDAP (or wider
+network) permissions should be granted for this user because you need
+to specify the service account password in the configuration file. A
+suitably strong password should be chosen, eg. VmLYBbvJaf2kAqcrt5HjHdG6
 
 
-### Taiga Front
+### taiga-front
 
-Change in your dist/js/conf.json the loginFormType setting to "ldap":
+Change the `loginFormType` setting to `"ldap"` in `dist/js/conf.json`:
 
 ```json
-...
+    ...
     "loginFormType": "ldap",
-...
+    ...
 ```

taiga_contrib_ldap_auth/__init__.py

@@ -14,6 +14,6 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
-__version__ = (0, 1, 1)
+__version__ = (0, 2, 0)
 
 default_app_config = "taiga_contrib_ldap_auth.apps.TaigaContribLDAPAuthAppConfig"

taiga_contrib_ldap_auth/connector.py

@@ -29,58 +29,84 @@ class LDAPLoginError(ConnectorBaseException):
 PORT = getattr(settings, "LDAP_PORT", "")
 
 SEARCH_BASE = getattr(settings, "LDAP_SEARCH_BASE", "")
-SEARCH_PROPERTY = getattr(settings, "LDAP_SEARCH_PROPERTY", "")
-SEARCH_SUFFIX = getattr(settings, "LDAP_SEARCH_SUFFIX", "")
-SEARCH_FILTER = getattr(settings, "LDAP_SEARCH_FILTER", "")
+SEARCH_FILTER_ADDITIONAL = getattr(settings, "LDAP_SEARCH_FILTER_ADDITIONAL", "")
 BIND_DN = getattr(settings, "LDAP_BIND_DN", "")
 BIND_PASSWORD = getattr(settings, "LDAP_BIND_PASSWORD", "")
 
-EMAIL_PROPERTY = getattr(settings, "LDAP_EMAIL_PROPERTY", "")
-FULL_NAME_PROPERTY = getattr(settings, "LDAP_FULL_NAME_PROPERTY", "")
+USERNAME_ATTRIBUTE = getattr(settings, "LDAP_USERNAME_ATTRIBUTE", "")
+EMAIL_ATTRIBUTE = getattr(settings, "LDAP_EMAIL_ATTRIBUTE", "")
+FULL_NAME_ATTRIBUTE = getattr(settings, "LDAP_FULL_NAME_ATTRIBUTE", "")
 
-def login(username: str, password: str) -> tuple:
+def login(login: str, password: str) -> tuple:
+    """
+    Connect to LDAP server, perform a search and attempt a bind.
 
-    try:
-        if SERVER.lower().startswith("ldaps://"):
-            server = Server(SERVER, port = PORT, get_info = NONE, use_ssl = True) 
-        else:
-            server = Server(SERVER, port = PORT, get_info = NONE, use_ssl = False)  # define an unsecure LDAP server, requesting info on DSE and schema
+    Can raise `exc.LDAPLoginError` exceptions if any of the
+    operations fail.
 
-        c = None
+    :returns: tuple (username, email, full_name)
 
-        if BIND_DN is not None and BIND_DN != '':
-            c = Connection(server, auto_bind = True, client_strategy = SYNC, user=BIND_DN, password=BIND_PASSWORD, authentication=SIMPLE, check_names=True)
-        else:
-            c = Connection(server, auto_bind = True, client_strategy = SYNC, user=None, password=None, authentication=ANONYMOUS, check_names=True)
+    """
 
+    # connect to the LDAP server
+    if SERVER.lower().startswith("ldaps://"):
+        use_ssl = True
+    else:
+        use_ssl = False
+    try:
+        server = Server(SERVER, port = PORT, get_info = NONE, use_ssl = use_ssl)
+    except Exception as e:
+        error = "Error connecting to LDAP server: %s" % e
+        raise LDAPLoginError({"error_message": error})
+
+    # authenticate as service if credentials provided, anonymously otherwise
+    if BIND_DN is not None and BIND_DN != '':
+        service_user = BIND_DN
+        service_pass = BIND_PASSWORD
+        service_auth = SIMPLE
+    else:
+        service_user = None
+        service_pass = None
+        service_auth = ANONYMOUS
+    try:
+        c = Connection(server, auto_bind = True, client_strategy = SYNC, check_names = True,
+                       user = service_user, password = service_pass, authentication = service_auth)
     except Exception as e:
         error = "Error connecting to LDAP server: %s" % e
         raise LDAPLoginError({"error_message": error})
 
+    # search for user-provided login
+    search_filter = '(|(%s=%s)(%s=%s))' % (USERNAME_ATTRIBUTE, login, EMAIL_ATTRIBUTE, login)
+    if SEARCH_FILTER_ADDITIONAL:
+        search_filter = '(&%s%s)' % (search_filter, SEARCH_FILTER_ADDITIONAL)
     try:
-        if(SEARCH_SUFFIX is not None and SEARCH_SUFFIX != ''):
-            search_filter = '(%s=%s)' % (SEARCH_PROPERTY, username + SEARCH_SUFFIX)
-        else:
-            search_filter = '(%s=%s)' % (SEARCH_PROPERTY, username)
-        if SEARCH_FILTER:
-            search_filter = '(&%s(%s))' % (search_filter, SEARCH_FILTER)
         c.search(search_base = SEARCH_BASE,
                  search_filter = search_filter,
                  search_scope = SUBTREE,
-                 attributes = [EMAIL_PROPERTY,FULL_NAME_PROPERTY],
+                 attributes = [USERNAME_ATTRIBUTE, EMAIL_ATTRIBUTE, FULL_NAME_ATTRIBUTE],
                  paged_size = 5)
+    except Exception as e:
+        error = "LDAP login incorrect: %s" % e
+        raise LDAPLoginError({"error_message": error})
 
-        if len(c.response) > 0:
-            dn = c.response[0].get('dn')
-            user_email = c.response[0].get('raw_attributes').get(EMAIL_PROPERTY)[0].decode('utf-8')
-            full_name = c.response[0].get('raw_attributes').get(FULL_NAME_PROPERTY)[0].decode('utf-8')
-
-            user_conn = Connection(server, auto_bind = True, client_strategy = SYNC, user = dn, password = password, authentication = SIMPLE, check_names = True)
-
-            return (user_email, full_name)
-
-        raise LDAPLoginError({"error_message": "Username or password incorrect"})
+    # stop if no search results
+    # TODO: handle multiple matches
+    if len(c.response) == 0:
+        raise LDAPLoginError({"error_message": "LDAP login not found"})
 
+    # attempt LDAP bind
+    username = c.response[0].get('raw_attributes').get(USERNAME_ATTRIBUTE)[0].decode('utf-8')
+    email = c.response[0].get('raw_attributes').get(EMAIL_ATTRIBUTE)[0].decode('utf-8')
+    full_name = c.response[0].get('raw_attributes').get(FULL_NAME_ATTRIBUTE)[0].decode('utf-8')
+    try:
+        dn = c.response[0].get('dn')
+        user_conn = Connection(server, auto_bind = True, client_strategy = SYNC,
+                               check_names = True, authentication = SIMPLE,
+                               user = dn, password = password)
     except Exception as e:
-        error = "LDAP account or password incorrect: %s" % e
+        error = "LDAP bind failed: %s" % e
         raise LDAPLoginError({"error_message": error})
+
+    # LDAP binding successful, but some values might have changed, or
+    # this is the user's first login, so return them
+    return (username, email, full_name)

taiga_contrib_ldap_auth/services.py

@@ -29,32 +29,37 @@ def ldap_register(username: str, email: str, full_name: str):
     """
     Register a new user from LDAP.
 
-    This can raise `exc.IntegrityError` exceptions in
+    Can raise `exc.IntegrityError` exceptions in
     case of conflict found.
 
     :returns: User
     """
     user_model = apps.get_model("users", "User")
 
     try:
-        # LDAP user association exist?
-        user = user_model.objects.get(username=username)
+        # LDAP user association exists?
+        user = user_model.objects.get(username = username)
     except user_model.DoesNotExist:
         # Create a new user
-        username_unique = slugify_uniquely(username, user_model, slugfield="username")
-        user = user_model.objects.create(email=email,
-                                         username=username_unique,
-                                         full_name=full_name)
-        user_registered_signal.send(sender=user.__class__, user=user)
+        username_unique = slugify_uniquely(username, user_model, slugfield = "username")
+        user = user_model.objects.create(username = username_unique,
+                                         email = email,
+                                         full_name = full_name)
+        user_registered_signal.send(sender = user.__class__, user = user)
 
     return user
 
 
 def ldap_login_func(request):
-    username = request.DATA.get('username', None)
-    password = request.DATA.get('password', None)
+    # although the form field is called 'username', it can be an e-mail
+    # (or any other attribute)
+    login_input = request.DATA.get('username', None)
+    password_input = request.DATA.get('password', None)
+
+    # TODO: make sure these fields are sanitized before passing to LDAP server!
+    username, email, full_name = connector.login(login = login_input, password = password_input)
+
+    user = ldap_register(username = username, email = email, full_name = full_name)
 
-    email, full_name = connector.login(username=username, password=password)
-    user = ldap_register(username=username, email=email, full_name=full_name)
     data = make_auth_response_data(user)
     return data

tests/test_ldap.py

@@ -31,18 +31,19 @@ def test_ldap_login_success():
         m_server.return_value = Mock()
         m_connection.return_value = Mock()
 
-        username = "**userName**"
+        login = "**userName**"
         password = "**password**"
-        (email, full_name) = connector.login(username, password)
-        assert email == username + BASE_EMAIL
-        assert full_name == username
+        (username, email, full_name) = connector.login(login, password)
+        assert username == login
+        assert email == login + BASE_EMAIL
+        assert full_name == login
 
 
 def test_ldap_login_fail():
     with pytest.raises(connector.LDAPLoginError) as e:
-        username = "**userName**"
+        login = "**userName**"
         password = "**password**"
-        auth_info = connector.login(username, password)
+        auth_info = connector.login(login, password)
 
     assert e.value.status_code == 400
     assert "error_message" in e.value.detail