fix(auth): sanitise username before LDAP DN and filter insertion

app/service/login_handlers/default.py

@@ -4,6 +4,8 @@
 from aiohttp import web
 from aiohttp_jinja2 import render_template
 from ldap3.core.exceptions import LDAPAttributeError, LDAPException
+from ldap3.utils.conv import escape_filter_chars
+from ldap3.utils.dn import escape_rdn_value
 
 from app.service.interfaces.i_login_handler import LoginHandlerInterface
 from app.utility.config_util import verify_hash
@@ -59,10 +61,11 @@ async def _ldap_login(self, username, password):
         dn = self._ldap_config.get('dn')
         user_attr = self._ldap_config.get('user_attr') or 'uid'
         user_format_string = self._ldap_config.get("user_format") or "{user_attr}={user},{dn}"
+        safe_username = escape_rdn_value(username)
         try:
-            user_string = user_format_string.format(user_attr=user_attr, user=username, dn=dn)
+            user_string = user_format_string.format(user_attr=user_attr, user=safe_username, dn=dn)
         except KeyError:
-            user_string = '%s=%s,%s' % (user_attr, username, dn)
+            user_string = '%s=%s,%s' % (user_attr, safe_username, dn)
 
         try:
             with ldap3.Connection(server, user=user_string, password=password) as conn:
@@ -82,7 +85,7 @@ async def _ldap_get_group(self, connection, dn, username, user_attr):
         red_group_name = self._ldap_config.get('red_group') or 'red'
 
         try:
-            connection.search(dn, '(%s=%s)' % (user_attr, username), attributes=[group_attr])
+            connection.search(dn, '(%s=%s)' % (user_attr, escape_filter_chars(username)), attributes=[group_attr])
         except LDAPAttributeError:
             self.log.error('Invalid group_attr in config: %s', group_attr)
             return 'blue'

tests/services/test_ldap_injection.py

@@ -0,0 +1,61 @@
+"""Tests that LDAP injection characters are properly escaped in the
+DefaultLoginHandler before they are inserted into DN strings or
+LDAP search filters."""
+import pytest
+
+from ldap3.utils.conv import escape_filter_chars
+from ldap3.utils.dn import escape_rdn_value
+
+
+class TestLdapRdnEscape:
+    """escape_rdn_value must neutralise DN injection characters."""
+
+    def test_escapes_comma(self):
+        result = escape_rdn_value("admin,dc=evil,dc=com")
+        assert "," not in result or result.startswith("\\")
+
+    def test_escapes_equals(self):
+        result = escape_rdn_value("user=injected")
+        # ldap3 escapes the = so it cannot break the DN structure
+        assert result != "user=injected"
+
+    def test_clean_username_unchanged(self):
+        result = escape_rdn_value("jsmith")
+        assert result == "jsmith"
+
+    def test_escapes_null_byte(self):
+        result = escape_rdn_value("admin\x00")
+        assert "\x00" not in result
+
+
+class TestLdapFilterEscape:
+    """escape_filter_chars must neutralise filter injection characters."""
+
+    def test_escapes_asterisk(self):
+        result = escape_filter_chars("*")
+        assert result == "\\2a"
+
+    def test_escapes_parentheses(self):
+        result = escape_filter_chars(")(uid=*")
+        assert "(" not in result
+        assert ")" not in result
+
+    def test_clean_username_unchanged(self):
+        result = escape_filter_chars("jsmith")
+        assert result == "jsmith"
+
+    def test_escapes_null_byte(self):
+        result = escape_filter_chars("admin\x00")
+        assert "\x00" not in result
+
+
+class TestDefaultLoginHandlerImports:
+    """Verify that the login handler imports the sanitisation helpers."""
+
+    def test_escape_rdn_value_imported(self):
+        from app.service.login_handlers.default import escape_rdn_value as _erv
+        assert callable(_erv)
+
+    def test_escape_filter_chars_imported(self):
+        from app.service.login_handlers.default import escape_filter_chars as _efc
+        assert callable(_efc)