ldap auth improvements

Author: nicozanf

apps/_scaffold/settings.py

@@ -104,6 +104,15 @@
     "mode": "ad",  # Microsoft Active Directory
     "server": "mydc.domain.com",  # FQDN or IP of one Domain Controller
     "base_dn": "cn=Users,dc=domain,dc=com",  # base dn, i.e. where the users are located
+
+    # the following are only needed if you want to use groups
+    "group_member_attrib": "member",  # for AD, attribute that contains the user DN in the group
+    "bind_dn": "CN=LdapBindUser,CN=users,DC=test,DC=local",  # bind user DN
+    "bind_pw": "P@ssw0rd",  # bind user password
+    "group_dn": "DC=test,DC=local",  # group DN, where the groups are located, default = base_dn
+    "allowed_groups": ["allowed_login_group"],  # list of groups that are allowed to log in, default = everyone
+    "denied_login_popup": True,  # show an error at login if not in allowed groups
+                                 # default = False
 }
 
 # i18n settings

py4web/utils/auth.py

@@ -702,7 +702,16 @@ def login(self, email, password):
                     user_info["sso_id"] = plugin.name + ":" + email
                     if self.use_username or "@" not in email:
                         user_info["username"] = email
-                    if "@" in email:
+                    # --- LDAP/AD: use real email if available ---
+                    if (
+                        plugin.name == "ldap"
+                        and getattr(plugin, "mode", None) == "ad"
+                        and getattr(plugin, "last_user_mail", None)
+                    ):
+                        self.logger.debug(f"Using AD email from LDAP plugin: {plugin.last_user_mail}")
+                        user_info["email"] = plugin.last_user_mail
+                    elif "@" in email:
+                        self.logger.debug(f"Using email from login: {email}")
                         user_info["email"] = email
                     else:
                         self.logger.debug(
@@ -1303,11 +1312,11 @@ def login(auth):
         if "pam" in auth.plugins or "ldap" in auth.plugins:
             plugin_name = "pam" if "pam" in auth.plugins else "ldap"
             plugin = auth.plugins[plugin_name]
-            self.logger.debug(
+            auth.logger.debug(
                 f"AuthAPI.login: Trying plugin {plugin_name} for user {username}"
             )
             check = plugin.check_credentials(username, password)
-            self.logger.debug(
+            auth.logger.debug(
                 f"AuthAPI.login: plugin.check_credentials returned {check}"
             )
             if check:
@@ -1316,19 +1325,29 @@ def login(auth):
                     # "email": username + "@localhost",
                     "sso_id": plugin_name + ":" + username,
                 }
-                # and register the user if we have one, just in case
+                # For AD, use the real email if available
+                if (
+                    plugin_name == "ldap"
+                    and getattr(plugin, "mode", None) == "ad"
+                    and getattr(plugin, "last_user_mail", None)
+                ):
+                    auth.logger.debug(f"AuthAPI.login: Using AD email from LDAP plugin: {plugin.last_user_mail}")
+                    # save the real email from AD to database
+                    data["email"] = plugin.last_user_mail
+                else:
+                    auth.logger.debug(f"AuthAPI.login: Not using AD email, plugin_name={plugin_name}, mode={getattr(plugin, 'mode', None)}, last_user_mail={getattr(plugin, 'last_user_mail', None)}")
                 if auth.db:
-                    self.logger.debug(
+                    auth.logger.debug(
                         f"AuthAPI.login: Calling get_or_register_user with data={data}"
                     )
                     user = auth.get_or_register_user(data)
-                    self.logger.debug(
+                    auth.logger.debug(
                         f"AuthAPI.login: User after get_or_register_user: {user}"
                     )
                     auth.store_user_in_session(user["id"])
                 # else: if we're here - check is OK, but user is not in the session - is it right?
             else:
-                self.logger.debug(
+                auth.logger.debug(
                     f"AuthAPI.login: plugin.check_credentials failed for {username}"
                 )
                 data = auth._error(

py4web/utils/auth_plugins/ldap_plugin.py

@@ -6,8 +6,9 @@
 import logging
 import sys
 
-import ldap  # python-ldap
-import ldap.filter
+import ldap  # python-ldap # type: ignore  # pylance: ignore undefined
+import ldap.filter  # type: ignore  # pylance: ignore undefined
+from py4web import HTTP
 
 from . import UsernamePassword
 
@@ -90,7 +91,7 @@ class LDAPPlugin(UsernamePassword):
 
     Where:
     manage_user: bool
-        If True py4web will fetch and update user profile
+        If True or AD is used, py4web will fetch and update user profile
         fields (first name, last name, email) from LDAP/AD on each login and
         keep them in sync with its db.
         If False, only authentication is performed and user profile fields
@@ -126,7 +127,6 @@ class LDAPPlugin(UsernamePassword):
         group_member_attrib - the attribute containing the group members name
         group_filterstr - as the filterstr but for group select
 
-    **allowed_groups still to be implemented in py4web**
     You can restrict login access to specific groups if you specify:
 
         auth.register_plugin(LDAPPlugin(
@@ -169,6 +169,7 @@ def __init__(
         username_attrib="uid",
         custom_scope="subtree",
         allowed_groups=None,
+        denied_login_popup=False,
         manage_user=False,
         user_firstname_attrib="cn:1",
         user_lastname_attrib="cn:2",
@@ -202,6 +203,7 @@ def __init__(
         self.username_attrib = username_attrib
         self.custom_scope = custom_scope
         self.allowed_groups = allowed_groups
+        self.denied_login_popup = denied_login_popup
         self.manage_user = manage_user
         self.user_firstname_attrib = user_firstname_attrib
         self.user_lastname_attrib = user_lastname_attrib
@@ -219,6 +221,7 @@ def __init__(
         # rfc4515 syntax
         self.filterstr = self.filterstr.lstrip("(").rstrip(")")
         self.groups = groups
+        self.last_user_mail = None
 
     def check_credentials(self, username, password):
         base_dn = self.base_dn
@@ -266,11 +269,14 @@ def check_credentials(self, username, password):
             user_lastname_attrib = ldap.filter.escape_filter_chars(user_lastname_attrib)
             user_mail_attrib = ldap.filter.escape_filter_chars(user_mail_attrib)
         try:
-            # if allowed_groups:
-            #    if not self.is_user_in_allowed_groups(
-            #        username, password, allowed_groups
-            #    ):
-            #        return False
+            allowed_groups = self.allowed_groups
+            if allowed_groups and mode == "ad":
+                if not self.is_user_in_allowed_groups(
+                    username
+                ):
+                    logger.warning(f"[{username}] refused login because not in allowed groups!")
+                    
+                    return False
             con = self._init_ldap()
             if mode == "ad":
                 # Microsoft Active Directory
@@ -295,7 +301,7 @@ def check_credentials(self, username, password):
                 # this will throw an index error if the account is not found
                 # in the base_dn
                 requested_attrs = ["sAMAccountName"]
-                if manage_user:
+                if manage_user or mode == "ad":
                     requested_attrs.extend(
                         [user_firstname_attrib, user_lastname_attrib, user_mail_attrib]
                     )
@@ -306,7 +312,14 @@ def check_credentials(self, username, password):
                     f"(&(sAMAccountName={ldap.filter.escape_filter_chars(username_bare)})({filterstr}))",
                     requested_attrs,
                 )[0][1]
+
+                # set last_user_mail from ldap result for further use
+                if 'mail' in result and result['mail']:
+                    self.last_user_mail = str(result['mail'][0].decode() if isinstance(result['mail'][0], bytes) else result['mail'][0])
+                else:
+                    self.last_user_mail = None
                 logger.info(f"Login result: {result}")
+                logger.debug(f"LDAPPlugin: Set last_user_mail to {self.last_user_mail} for {username}")
                 if not isinstance(result, dict):
                     # result should be a dict in the form
                     # {'sAMAccountName': [username_bare]}
@@ -533,6 +546,8 @@ def check_credentials(self, username, password):
 
     def is_user_in_allowed_groups(self, username, password=None):
         allowed_groups = self.allowed_groups
+        logger = self.logger
+        logger.debug(f"[{str(username)}] Check if user is in allowed groups")
 
         """
         Figure out if the username is a member of an allowed group
@@ -549,8 +564,12 @@ def is_user_in_allowed_groups(self, username, password=None):
         for group in allowed_groups:
             if group in self.groups:
                 # Match
+                logger.debug(f"[{str(username)}] allowed login because in allowed groups")
                 return True
         # No match
+        logger.warning(f"[{str(username)}] denied login because not in allowed groups")
+        if self.denied_login_popup:
+            raise HTTP(401, body=f"{str(username)} : you're not authorized to login!")
         return False
 
     def do_manage_groups(self, con, username, group_mapping={}):
@@ -729,6 +748,8 @@ def get_user_groups_from_ldap(self, username=None, password=None):
                     if "DC=" in x.upper():
                         domain.append(x.split("=")[-1])
                 username = f"{username}@{'.'.join(domain)}"
+            if not group_dn:
+                group_dn = base_dn
             username_bare = username.split("@")[0]
             con = self._init_ldap()
             con.set_option(ldap.OPT_PROTOCOL_VERSION, 3)
@@ -758,6 +779,11 @@ def get_user_groups_from_ldap(self, username=None, password=None):
         if username is None:
             return []
         # search for groups where user is in
+        filter_full = (
+            ldap.filter.escape_filter_chars(group_member_attrib),
+            ldap.filter.escape_filter_chars(username),
+            group_filterstr,
+        )
         filter = f"(&({ldap.filter.escape_filter_chars(group_member_attrib)}=\
             {ldap.filter.escape_filter_chars(username)})({group_filterstr}))"