feat(ldap): add a field that allows to override LDAP User Object Filter when a user is imported

cps/admin.py

@@ -564,6 +564,7 @@ def _configuration_ldap_helper(to_save, gdriveError):
     reboot_required |= _config_string(to_save, "config_ldap_user_object")
     reboot_required |= _config_string(to_save, "config_ldap_group_object_filter")
     reboot_required |= _config_string(to_save, "config_ldap_group_members_field")
+    reboot_required |= _config_string(to_save, "config_ldap_member_user_object")
     reboot_required |= _config_checkbox(to_save, "config_ldap_openldap")
     reboot_required |= _config_int(to_save, "config_ldap_encryption")
     reboot_required |= _config_string(to_save, "config_ldap_cert_path")
@@ -598,10 +599,17 @@ def _configuration_ldap_helper(to_save, gdriveError):
 
     if config.config_ldap_user_object.count("%s") != 1:
         return reboot_required, _configuration_result(_('LDAP User Object Filter needs to Have One "%s" Format Identifier'),
-                                     gdriveError)
+                                                      gdriveError)
     if config.config_ldap_user_object.count("(") != config.config_ldap_user_object.count(")"):
         return reboot_required, _configuration_result(_('LDAP User Object Filter Has Unmatched Parenthesis'),
-                                     gdriveError)
+                                                      gdriveError)
+
+    if config.config_ldap_member_user_object.count("%s") != 1:
+        return reboot_required, _configuration_result(_('LDAP Member User Filter needs to Have One "%s" Format Identifier'),
+                                                      gdriveError)
+    if config.config_ldap_member_user_object.count("(") != config.config_ldap_member_user_object.count(")"):
+        return reboot_required, _configuration_result(_('LDAP Member User Filter Has Unmatched Parenthesis'),
+                                                      gdriveError)
 
     if config.config_ldap_cert_path and not os.path.isdir(config.config_ldap_cert_path):
         return reboot_required, _configuration_result(_('LDAP Certificate Location is not Valid, Please Enter Correct Path'),

cps/config_sql.py

@@ -112,6 +112,7 @@ class _Settings(_Base):
     config_ldap_cert_path = Column(String, default="")
     config_ldap_dn = Column(String, default='dc=example,dc=org')
     config_ldap_user_object = Column(String, default='uid=%s')
+    config_ldap_member_user_object = Column(String, default='cn=%s')
     config_ldap_openldap = Column(Boolean, default=True)
     config_ldap_group_object_filter = Column(String, default='(&(objectclass=posixGroup)(cn=%s))')
     config_ldap_group_members_field = Column(String, default='memberUid')

cps/services/simpleldap.py

@@ -64,6 +64,7 @@ def init_app(app, config):
     app.config['LDAP_OPENLDAP'] = bool(config.config_ldap_openldap)
     app.config['LDAP_GROUP_OBJECT_FILTER'] = config.config_ldap_group_object_filter
     app.config['LDAP_GROUP_MEMBERS_FIELD'] = config.config_ldap_group_members_field
+    app.config['LDAP_MEMBER_USER_OBJECT_FILTER'] = config.config_ldap_member_user_object
 
     _ldap.init_app(app)
 

cps/templates/config_edit.html

@@ -310,6 +310,10 @@
             <label for="config_ldap_group_members_field">{{_('LDAP Group Members Field')}}</label>
             <input type="text" class="form-control" id="config_ldap_group_members_field" name="config_ldap_group_members_field" value="{% if config.config_ldap_group_members_field != None %}{{ config.config_ldap_group_members_field }}{% endif %}" autocomplete="off">
           </div>
+          <div class="form-group">
+              <label for="config_ldap_user_object">{{_('LDAP Member User Filter')}}</label>
+              <input type="text" class="form-control" id="config_ldap_member_user_object" name="config_ldap_member_user_object" value="{% if config.config_ldap_member_user_object != None %}{{ config.config_ldap_member_user_object }}{% endif %}" autocomplete="off">
+          </div>
       </div>
       {% endif %}
       {% if feature_support['oauth'] %}

cps/web.py

@@ -319,28 +319,30 @@ def import_ldap_users():
     for username in new_users:
         user = username.decode('utf-8')
         if '=' in user:
-            match = re.search("([a-zA-Z0-9-]+)=%s", config.config_ldap_user_object, re.IGNORECASE | re.UNICODE)
-            if match:
-                match_filter = match.group(1)
-                match = re.search(match_filter + "=([\d\s\w-]+)", user, re.IGNORECASE | re.UNICODE)
-                if match:
-                    user = match.group(1)
-                else:
-                    log.warning("Could Not Parse LDAP User: %s", user)
-                    continue
+            if config.config_ldap_member_user_object:
+                query_filter = config.config_ldap_member_user_object
             else:
-                log.warning("Could Not Parse LDAP User: %s", user)
+                query_filter = config.config_ldap_user_object
+
+            try:
+                user_identifier = extract_user_identifier_from_ldap_with_filter(user, query_filter)
+            except Exception as e:
+                log.warning(e)
                 continue
-        if ub.session.query(ub.User).filter(ub.User.nickname == user.lower()).first():
-            log.warning("LDAP User: %s Already in Database", user)
+        else:
+            user_identifier = user
+
+        if ub.session.query(ub.User).filter(ub.User.nickname == user_identifier.lower()).first():
+            log.warning("LDAP User: %s Already in Database", user_identifier)
             continue
-        user_data = services.ldap.get_object_details(user=user,
+        user_data = services.ldap.get_object_details(user=user_identifier,
                                                      group=None,
-                                                     query_filter=None,
+                                                     query_filter=query_filter,
                                                      dn_only=False)
         if user_data:
             content = ub.User()
-            content.nickname = user
+            user_login_field = extract_dynamic_field_from_filter(user, config.config_ldap_user_object)
+            content.nickname = user_data[user_login_field][0].decode('utf-8')
             content.password = ''  # dummy password which will be replaced by ldap one
             if 'mail' in user_data:
                 content.email = user_data['mail'][0].decode('utf-8')
@@ -370,6 +372,27 @@ def import_ldap_users():
     return json.dumps(showtext)
 
 
+def extract_user_data_from_field(user, field):
+    match = re.search(field + "=([\d\s\w-]+)", user, re.IGNORECASE | re.UNICODE)
+    if match:
+        return match.group(1)
+    else:
+        raise Exception("Could Not Parse LDAP User: %s", user)
+
+
+def extract_dynamic_field_from_filter(user, filter):
+    match = re.search("([a-zA-Z0-9-]+)=%s", filter, re.IGNORECASE | re.UNICODE)
+    if match:
+        return match.group(1)
+    else:
+        raise Exception("Could Not Parse LDAP User: %s", user)
+
+
+def extract_user_identifier_from_ldap_with_filter(user, filter):
+    dynamic_field = extract_dynamic_field_from_filter(user, filter)
+    return extract_user_data_from_field(user, dynamic_field)
+
+
 # ################################### data provider functions #########################################################