feat: Allow binding with user-provided credentials

README.md

@@ -27,9 +27,6 @@ If you were to start Taiga now, it would not pull the `taiga-back` directly from
 
 #### `custom-back/Dockerfile`
 
-<details>
-<summary>Click here to expand</summary>
-
 ```Dockerfile
 FROM taigaio/taiga-back:latest
 
@@ -40,6 +37,9 @@ RUN cat /taiga-back/settings/config.append.py >> /taiga-back/settings/config.py
 RUN pip install taiga-contrib-ldap-auth-ext
 ```
 
+<details>
+<summary>Click here to expand explanation</summary>
+
 The statements in the Dockerfile have the following effect:
 
 1. `FROM ...` bases the image we build on the official `taigaio/taiga-back` image.
@@ -110,12 +110,9 @@ Change the `loginFormType` setting to `"ldap"` in `dist/conf.json`:
 
 ### `taiga-back` configuration
 
-If you use the installation with Docker, put the following contents in the file `custom-back/config.append.py`. 
-
-If you use the installation without Docker, append the following contents to the file `settings/common.py` (for Taiga >5.0) or `settings/local.py` (for Taiga ≤5.0). 
+If you use the installation with Docker, put something similar to the following in the file `custom-back/config.append.py`. 
 
-<details>
-<summary>Click here to expand</summary>
+If you use the installation without Docker, append something similar to the following to the file `settings/common.py` (for Taiga >5.0) or `settings/local.py` (for Taiga ≤5.0):
 
 ```python
 INSTALLED_APPS += ["taiga_contrib_ldap_auth_ext"]
@@ -139,7 +136,10 @@ LDAP_SAVE_LOGIN_PASSWORD = False
 LDAP_MAP_USERNAME_TO_UID = None
 ```
 
-Change the following fields matching your setup:
+_You need to change most of the values to match your setup._
+
+<details>
+<summary>Click here to expand configuration explanation</summary>
 
 **`LDAP_SERVER` and `LDAP_PORT`:** You will definitely have to change the server URL. If possible, try to keep the `ldaps://` to use a secure connection. The port can likely stay as is, unless...
 
@@ -151,7 +151,7 @@ Change the following fields matching your setup:
     LDAP_PORT = 389
     LDAP_START_TLS = True
     ```
-    What happens is that an unencrypted connection is established first, but then upgraded to a secure connection. To the best of my knowledge, this should also be safe – however, I like the `ldaps://` variant more.
+    What happens is that an unencrypted connection is established first, but then upgraded to a secure connection. This is [less secure](https://docs.redhat.com/de/documentation/red_hat_directory_server/12/html/securing_red_hat_directory_server/assembly_enabling-tls-encrypted-connections-to-directory-server_securing-rhds#assembly_enabling-tls-encrypted-connections-to-directory-server_securing-rhds) than `ldaps://` (see also [the related discussion for STARTTLS for emails](https://serverfault.com/questions/523804/is-starttls-less-safe-than-tls-ssl) or [this blog post](https://blog.apnic.net/2021/11/18/vulnerabilities-show-why-starttls-should-be-avoided-if-possible/)), because an attacker could strip the “upgrade to secure connection” request causing the connection to remain insecure. It is still safer than an unecrypted connection, of course.
 
 **`LDAP_BIND_DN`, `LDAP_BIND_PASSWORD`**: You will need to change them. 
 
@@ -163,6 +163,14 @@ If `LDAP_BIND_DN` is not specified or blank, an anonymous bind is attempted.
 
 It is recommended to limit the service account and only allow it to read and search the LDAP structure (no write or other LDAP access). The credentials should also not be used for any other account on the network. This minimizes the damage in cases of a successful LDAP injection or if you ever accidentially give someone access to the configuration file (e.g. by committing it into version control or having misconfigured permissions). Use a suitably strong, ideally randomly generated password.
 
+You can also use the credentials provided by the user to bind to LDAP (eliminating the need for a dedicated LDAP service account). To do so, do the following three things:
+
+1. Set `LDAP_BIND_WITH_USER_PROVIDED_CREDENTIALS = True`
+2. Insert the placeholder `<username>` inside `LDAP_BIND_DN`, e.g. like this: `"CN=<username>,OU=DevTeam,DC=example,DC=com"`.
+3. Remove `LDAP_BIND_PASSWORD` (it will not be used)
+
+Taiga will then determine the LDAP bind user by replacing `<username>` with the user-provided username, and bind using the user-provided password.
+
 **`LDAP_SEARCH_BASE`**: The subtree where the users are located.
 
 **`LDAP_USERNAME_ATTRIBUTE`, `LDAP_EMAIL_ATTRIBUTE`, `LDAP_FULL_NAME_ATTRIBUTE`**: These are the LDAP attributes used to get the username, email and full name shown in the Taiga application. They need to have a value in LDAP. Depending on your LDAP setup, you might need to change them.
@@ -177,7 +185,7 @@ It is recommended to limit the service account and only allow it to read and sea
 #### Additional configuration options
 
 <details>
-<summary>Click here to expand</summary>
+<summary>Click here to expand additional configuration options</summary>
 
 By default, Taiga will fall back to `normal` authentication if LDAP authentication fails. Add the following line to disable this and only allow LDAP login:
 

taiga_contrib_ldap_auth_ext/connector.py

@@ -11,7 +11,19 @@
 # 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/>.
 
-from ldap3 import Server, Connection, AUTO_BIND_NO_TLS, AUTO_BIND_TLS_BEFORE_BIND, ANONYMOUS, SIMPLE, SYNC, SUBTREE, NONE
+from typing import Any, Dict, Tuple
+
+from ldap3 import (
+    Server,
+    Connection,
+    AUTO_BIND_NO_TLS,
+    AUTO_BIND_TLS_BEFORE_BIND,
+    ANONYMOUS,
+    SIMPLE,
+    SYNC,
+    SUBTREE,
+    NONE,
+)
 
 from django.conf import settings
 from ldap3.utils.conv import escape_filter_chars
@@ -35,20 +47,115 @@ class LDAPUserLoginError(LDAPError):
 PORT = getattr(settings, "LDAP_PORT", "389")
 
 SEARCH_BASE = getattr(settings, "LDAP_SEARCH_BASE", "")
-SEARCH_FILTER_ADDITIONAL = getattr(
-    settings, "LDAP_SEARCH_FILTER_ADDITIONAL", "")
+SEARCH_FILTER_ADDITIONAL = getattr(settings, "LDAP_SEARCH_FILTER_ADDITIONAL", "")
 BIND_DN = getattr(settings, "LDAP_BIND_DN", "")
+BIND_WITH_USER_PROVIDED_CREDENTIALS = getattr(
+    settings, "LDAP_BIND_WITH_USER_PROVIDED_CREDENTIALS", False
+)
 BIND_PASSWORD = getattr(settings, "LDAP_BIND_PASSWORD", "")
 
 USERNAME_ATTRIBUTE = getattr(settings, "LDAP_USERNAME_ATTRIBUTE", "uid")
 EMAIL_ATTRIBUTE = getattr(settings, "LDAP_EMAIL_ATTRIBUTE", "mail")
 FULL_NAME_ATTRIBUTE = getattr(settings, "LDAP_FULL_NAME_ATTRIBUTE", "displayName")
+PROFILE_ATTRIBUTES = [USERNAME_ATTRIBUTE, EMAIL_ATTRIBUTE, FULL_NAME_ATTRIBUTE]
 
 TLS_CERTS = getattr(settings, "LDAP_TLS_CERTS", "")
 START_TLS = getattr(settings, "LDAP_START_TLS", False)
 
 
-def login(username: str, password: str) -> tuple:
+def _get_server() -> Server:
+    """Connect to an LDAP server (no authentication yet)."""
+    tls = TLS_CERTS or None
+    use_ssl = SERVER.lower().startswith("ldaps://")
+
+    try:
+        return Server(SERVER, port=PORT, get_info=NONE, use_ssl=use_ssl, tls=tls)
+    except Exception as e:
+        error = f"Error connecting to LDAP server: {e}"
+        raise LDAPConnectionError({"error_message": error})
+
+
+def _get_auth_details(
+    username_sanitized: str, user_provided_password: str
+) -> Dict[str, Any]:
+    """
+    Return a dictionary with LDAP auth credentials.
+
+    The dictionary contains the following fields:
+
+    - "user": DN of the user to bind with
+    - "password": Password of the user to bind with
+    - "authentication": Bind method
+
+    The bind method may be SIMPLE or ANONYMOUS.
+    The user to bind with may be a dedicated bind user, or a dynamically
+    determined DN from the provided user credentials.
+    """
+    if BIND_WITH_USER_PROVIDED_CREDENTIALS:
+        # Authenticate using the provided user credentials
+        user = BIND_DN.replace("<username>", username_sanitized)
+        password = user_provided_password
+        authentication = SIMPLE
+    elif BIND_DN:
+        # Authenticate with dedicated bind credentials
+        user = BIND_DN
+        password = BIND_PASSWORD
+        authentication = SIMPLE
+    else:
+        # Use anonymous auth
+        user = None
+        password = None
+        authentication = ANONYMOUS
+
+    return {"user": user, "password": password, "authentication": authentication}
+
+
+def _extract_user(response: Any) -> Any:
+    """
+    Extract a single user object from the LDAP response.
+
+    Throw an error if there is not exactly 1 user in the response.
+    """
+    users_found = [r for r in response if "raw_attributes" in r and "dn" in r]
+
+    # stop if no search results
+    if not users_found:
+        raise LDAPUserLoginError({"error_message": "LDAP login not found"})
+
+    # handle multiple matches
+    if len(users_found) > 1:
+        raise LDAPUserLoginError(
+            {"error_message": "LDAP login could not be determined."}
+        )
+
+    return users_found[0]
+
+
+def _extract_profile(user: Any) -> Tuple[str, str, str]:
+    """
+    Extract the profile from the given user.
+
+    The profile consists of the following attributes:
+
+    - Username
+    - Email
+    - Full name
+
+    Throw an error if the attributes are not all set.
+    """
+    raw_attributes = user.get("raw_attributes")
+
+    for attribute in PROFILE_ATTRIBUTES:
+        if not raw_attributes.get(attribute):
+            raise LDAPUserLoginError({"error_message": "LDAP login is invalid."})
+
+    return tuple(
+        raw_attributes.get(attribute)[0].decode("utf-8")
+        for attribute in PROFILE_ATTRIBUTES
+    )
+
+
+def login(username_or_email: str, password: str) -> Tuple[str, str, str]:
     """
     Connect to LDAP server, perform a search and attempt a bind.
 
@@ -58,97 +165,61 @@ def login(username: str, password: str) -> tuple:
     Can raise `exc.LDAPUserLoginError` exceptions if the
     login to LDAP fails.
 
-    :param username: a possibly unsanitized username
+    :param username_or_email: a possibly unsanitized username or email
     :param password: a possibly unsanitized password
     :returns: tuple (username, email, full_name)
-
     """
-
-    tls = None
-    if TLS_CERTS:
-        tls = TLS_CERTS
-
-    # 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, tls=tls)
-    except Exception as e:
-        error = "Error connecting to LDAP server: %s" % e
-        raise LDAPConnectionError({"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
-
-    auto_bind = AUTO_BIND_NO_TLS
-    if START_TLS:
-        auto_bind = AUTO_BIND_TLS_BEFORE_BIND
+    server = _get_server()
+    username_or_email_sanitized = escape_filter_chars(username_or_email)
+    auto_bind = AUTO_BIND_TLS_BEFORE_BIND if START_TLS else AUTO_BIND_NO_TLS
 
     try:
-        c = Connection(server, auto_bind=auto_bind, client_strategy=SYNC, check_names=True,
-                       user=service_user, password=service_pass, authentication=service_auth)
+        c = Connection(
+            server,
+            auto_bind=auto_bind,
+            client_strategy=SYNC,
+            check_names=True,
+            **_get_auth_details(username_or_email_sanitized, password),
+        )
     except Exception as e:
-        error = "Error connecting to LDAP server: %s" % e
+        error = f"Error connecting to LDAP server: {e}"
         raise LDAPConnectionError({"error_message": error})
 
     # search for user-provided login
-    username_sanitized = escape_filter_chars(username)
-    search_filter = '(|(%s=%s)(%s=%s))' % (
-        USERNAME_ATTRIBUTE, username_sanitized, EMAIL_ATTRIBUTE, username_sanitized)
+    search_filter = f"(|({USERNAME_ATTRIBUTE}={username_or_email_sanitized})({EMAIL_ATTRIBUTE}={username_or_email_sanitized}))"
     if SEARCH_FILTER_ADDITIONAL:
-        search_filter = '(&%s%s)' % (search_filter, SEARCH_FILTER_ADDITIONAL)
+        search_filter = f"(&{search_filter}{SEARCH_FILTER_ADDITIONAL})"
     try:
-        c.search(search_base=SEARCH_BASE,
-                 search_filter=search_filter,
-                 search_scope=SUBTREE,
-                 attributes=[USERNAME_ATTRIBUTE,
-                             EMAIL_ATTRIBUTE, FULL_NAME_ATTRIBUTE],
-                 paged_size=5)
+        c.search(
+            search_base=SEARCH_BASE,
+            search_filter=search_filter,
+            search_scope=SUBTREE,
+            attributes=PROFILE_ATTRIBUTES,
+            paged_size=5,
+        )
     except Exception as e:
-        error = "LDAP login incorrect: %s" % e
+        error = f"LDAP login incorrect: {e}"
         raise LDAPUserLoginError({"error_message": error})
 
-    # we are only interested in user objects in the response
-    c.response = [r for r in c.response if 'raw_attributes' in r and 'dn' in r]
-    # stop if no search results
-    if not c.response:
-        raise LDAPUserLoginError({"error_message": "LDAP login not found"})
-
-    # handle multiple matches
-    if len(c.response) > 1:
-        raise LDAPUserLoginError(
-            {"error_message": "LDAP login could not be determined."})
-
-    # handle missing mandatory attributes
-    raw_attributes = c.response[0].get('raw_attributes')
-    if not (raw_attributes.get(USERNAME_ATTRIBUTE) and
-            raw_attributes.get(EMAIL_ATTRIBUTE) and
-            raw_attributes.get(FULL_NAME_ATTRIBUTE)):
-        raise LDAPUserLoginError({"error_message": "LDAP login is invalid."})
+    user = _extract_user(c.response)
+    user_profile = _extract_profile(user)
 
     # attempt LDAP bind
-    username = raw_attributes.get(USERNAME_ATTRIBUTE)[0].decode('utf-8')
-    email = raw_attributes.get(EMAIL_ATTRIBUTE)[0].decode('utf-8')
-    full_name = raw_attributes.get(FULL_NAME_ATTRIBUTE)[0].decode('utf-8')
     try:
-        dn = str(bytes(c.response[0].get('dn'), 'utf-8'), encoding='utf-8')
-        Connection(server, auto_bind=auto_bind, client_strategy=SYNC,
-                   check_names=True, authentication=SIMPLE,
-                   user=dn, password=password)
+        dn = str(bytes(user.get("dn"), "utf-8"), encoding="utf-8")
+        Connection(
+            server,
+            auto_bind=auto_bind,
+            client_strategy=SYNC,
+            check_names=True,
+            authentication=SIMPLE,
+            user=dn,
+            password=password,
+        )
     except Exception as e:
-        error = "LDAP bind failed: %s" % e
+        error = f"LDAP bind failed: {e}"
         raise LDAPUserLoginError({"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)
+    # Return user profile so that it can be used by Taiga,
+    # e.g., to set the user's full name in the database
+    return user_profile

taiga_contrib_ldap_auth_ext/services.py

@@ -57,7 +57,7 @@ def ldap_login_func(request):
 
     try:
         username, email, full_name = connector.login(
-            username=login_input, password=password_input)
+            username_or_email=login_input, password=password_input)
     except connector.LDAPUserLoginError as ldap_error:
         # If no fallback authentication is specified, raise the original LDAP error
         if not FALLBACK: