prevent ldap injection

Author: BNH440

apps/ldap/utils.py

@@ -24,6 +24,8 @@
     Connection,
     Server,
 )
+from ldap3.utils.conv import escape_filter_chars
+from ldap3.utils.dn import escape_rdn
 
 LDAP_SERVER_URL = "ldaps://ldap.csua.berkeley.edu"
 # TODO: make things faster because connect_timeout=2 was too slow (caused socket closure)
@@ -75,11 +77,12 @@ def make_password(password):
 
 
 def change_password(username, new_password):
+    sanitized_username = escape_rdn(username)
     # using newuser_connection for edit privileges
     with newuser_connection() as c:
         if c.bind():
             success = c.modify(
-                "uid={0},{1}".format(username, PEOPLE_OU),
+                "uid={0},{1}".format(sanitized_username, PEOPLE_OU),
                 {"userpassword": [MODIFY_REPLACE, make_password(new_password)]},
             )
             return success
@@ -95,9 +98,10 @@ def create_new_user(username, name, email, sid, password):
 
     If uid is -1, this means the bind failed.
     """
+    sanitized_username = escape_rdn(username)
     with newuser_connection() as c:
         if c.bind():
-            dn = "uid={0},{1}".format(username, PEOPLE_OU)
+            dn = "uid={0},{1}".format(sanitized_username, PEOPLE_OU)
             uid = get_max_uid() + 1
             attrs = {
                 "uid": username,
@@ -123,9 +127,10 @@ def delete_user(username):
     """Deletes a user. Returns True on successful deletion, False on failed
     deletion, and raises a RuntimeError if newuser fails to bind.
     """
+    sanitized_username = escape_rdn(username)
     with newuser_connection() as c:
         if c.bind():
-            dn = f"uid={username},{PEOPLE_OU}"
+            dn = f"uid={sanitized_username},{PEOPLE_OU}"
             success = c.delete(dn)
             return success
         else:
@@ -173,7 +178,8 @@ def authenticate(username, password):
     """
     verifies that the username and password are correct
     """
-    user_dn = "uid={0},{1}".format(username, PEOPLE_OU)
+    sanitized_username = escape_rdn(username)
+    user_dn = "uid={0},{1}".format(sanitized_username, PEOPLE_OU)
     with ldap_connection(user=user_dn, password=password) as c:
         if c.bind():
             return True
@@ -189,7 +195,8 @@ def get_all_groups():
 
 
 def get_group_members(group):
-    search_filter = "(cn={0})".format(group)
+    sanitized_group = escape_filter_chars(group)
+    search_filter = "(cn={0})".format(sanitized_group)
     with ldap_connection() as c:
         c.search(GROUP_OU, search_filter, attributes=ALL_ATTRIBUTES)
         if len(c.entries) == 0:
@@ -217,11 +224,12 @@ def get_politburo():
 
 
 def get_user_creation_time(username):
+    sanitized_username = escape_filter_chars(username)
     # WIP
     with ldap_connection() as c:
         c.search(
             PEOPLE_OU,
-            "(uid={0})".format(username),
+            "(uid={0})".format(sanitized_username),
             attributes=[
                 "createTimestamp",
                 "creatorsName",
@@ -241,35 +249,39 @@ def get_user_creation_time(username):
 
 
 def get_user_info(username):
+    sanitized_username = escape_filter_chars(username)
     with ldap_connection() as c:
-        c.search(PEOPLE_OU, "(uid={0})".format(username), attributes="*")
+        c.search(PEOPLE_OU, "(uid={0})".format(sanitized_username), attributes="*")
         if len(c.entries) == 0:
             raise Http404("No such user!")
 
         return c.entries[0]
 
 
 def get_user_gecos(username):
+    sanitized_username = escape_filter_chars(username)
     with ldap_connection() as c:
-        c.search(PEOPLE_OU, "(uid={0})".format(username), attributes="gecos")
+        c.search(PEOPLE_OU, "(uid={0})".format(sanitized_username), attributes="gecos")
         if len(c.entries) == 0:
             raise Http404("No such user!")
 
         return str(c.entries[0].gecos)
 
 
 def get_user_hashed_password(username):
+    sanitized_username = escape_filter_chars(username)
     with newuser_connection() as c:
-        c.search(PEOPLE_OU, "(uid={0})".format(username), attributes="userpassword")
+        c.search(PEOPLE_OU, "(uid={0})".format(sanitized_username), attributes="userpassword")
         if len(c.entries) == 0:
             raise Http404("No such user!")
 
         return str(c.entries[0].userpassword)
 
 
 def user_exists(username):
+    sanitized_username = escape_filter_chars(username)
     with ldap_connection() as c:
-        c.search(PEOPLE_OU, "(uid={0})".format(username), attributes="")
+        c.search(PEOPLE_OU, "(uid={0})".format(sanitized_username), attributes="")
         return len(c.entries) == 1
 
 
@@ -289,17 +301,19 @@ def get_user_email(username):
 
 
 def email_exists(email):
+    sanitized_email = escape_filter_chars(email)
     with ldap_connection() as c:
-        search_filter = "(gecos=*{0})".format(email)
+        search_filter = "(gecos=*{0})".format(sanitized_email)
         c.search(PEOPLE_OU, search_filter, attributes="gecos")
         if len(c.entries) > 0:
             return True
         return False
 
 
 def get_user_groups(username):
+    sanitized_username = escape_filter_chars(username)
     with ldap_connection() as c:
-        c.search(GROUP_OU, "(memberUid={})".format(username), attributes="cn")
+        c.search(GROUP_OU, "(memberUid={})".format(sanitized_username), attributes="cn")
         groups = [str(entry.cn) for entry in c.entries]
         return groups