Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions apps/_scaffold/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
33 changes: 26 additions & 7 deletions py4web/utils/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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:
Expand All @@ -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(
Expand Down
46 changes: 36 additions & 10 deletions py4web/utils/auth_plugins/ldap_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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]
)
Expand All @@ -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]}
Expand Down Expand Up @@ -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
Expand All @@ -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={}):
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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}))"

Expand Down
10 changes: 5 additions & 5 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.