dinislage/web2py
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ldap_auth.py pep8, thanks Kory

Browse Source
This commit is contained in:
mdipierro committed 2012-07-12 11:26:45 -05:00
1 parent d66d46be7e
commit d9802c374a
2 files changed
+76 -69
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

VERSION
+1 -1
View File
@@ -1 +1 @@
Version 2.00.0 (2012-07-11 23:13:56) dev Version 2.00.0 (2012-07-12 11:26:41) dev
gluon/contrib/login_methods/ldap_auth.py
+75 -68
View File
@@ -1,6 +1,6 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# #
# last tinkered with by korylprince at gmail.com on 2012-07-11 # last tinkered with by korylprince at gmail.com on 2012-07-12
# #
import sys import sys
@@ -13,6 +13,7 @@ except Exception, e:
logging.error('missing ldap, try "easy_install python-ldap"') logging.error('missing ldap, try "easy_install python-ldap"')
raise e raise e
def ldap_auth(server='ldap', port=None, def ldap_auth(server='ldap', port=None,
base_dn='ou=users,dc=domain,dc=com', base_dn='ou=users,dc=domain,dc=com',
mode='uid', secure=False, cert_path=None, cert_file=None, mode='uid', secure=False, cert_path=None, cert_file=None,
@@ -50,7 +51,8 @@ def ldap_auth(server = 'ldap', port = None,
auth.settings.login_methods.append(ldap_auth( auth.settings.login_methods.append(ldap_auth(
server='my.ldap.server', base_dn='ou=Users,dc=domain,dc=com')) server='my.ldap.server', base_dn='ou=Users,dc=domain,dc=com'))
to use ldap login with OpenLDAP and subtree search and (optionally) multiple DNs: to use ldap login with OpenLDAP and subtree search and (optionally)
multiple DNs:
auth.settings.login_methods.append(ldap_auth( auth.settings.login_methods.append(ldap_auth(
mode='uid_r', server='my.ldap.server', mode='uid_r', server='my.ldap.server',
@@ -73,16 +75,18 @@ def ldap_auth(server = 'ldap', port = None,
the custom_scope can be: base, onelevel, subtree. the custom_scope can be: base, onelevel, subtree.
If using secure ldaps:// pass secure=True and cert_path="..." If using secure ldaps:// pass secure=True and cert_path="..."
If ldap is using GnuTLS then you need cert_file="..." instead cert_path because If ldap is using GnuTLS then you need cert_file="..." instead cert_path
cert_path isn't implemented in GnuTLS :( because cert_path isn't implemented in GnuTLS :(
If you need to bind to the directory with an admin account in order to search it then specify bind_dn & bind_pw to use for this. If you need to bind to the directory with an admin account in order to
search it then specify bind_dn & bind_pw to use for this.
- currently only implemented for Active Directory - currently only implemented for Active Directory
If you need to restrict the set of allowed users (e.g. to members of a department) then specify If you need to restrict the set of allowed users (e.g. to members of a
a rfc4515 search filter string. department) then specify an rfc4515 search filter string.
- currently only implemented for mode in ['ad', 'company', 'uid_r'] - currently only implemented for mode in ['ad', 'company', 'uid_r']
You can manage user attribute first name, last name, email from ldap:
You can manage user attributes first name, last name, email from ldap:
auth.settings.login_methods.append(ldap_auth(...as usual..., auth.settings.login_methods.append(ldap_auth(...as usual...,
manage_user=True, manage_user=True,
user_firstname_attrib='cn:1', user_firstname_attrib='cn:1',
@@ -98,10 +102,11 @@ def ldap_auth(server = 'ldap', port = None,
user_lastname_attrib - the attribute containing the user's last name user_lastname_attrib - the attribute containing the user's last name
optionally you can specify parts. optionally you can specify parts.
Example: cn: "John Smith" - 'cn:2'='Smith' Example: cn: "John Smith" - 'cn:2'='Smith'
user_mail_attrib - the attribure containing the user's email address user_mail_attrib - the attribute containing the user's email address
If you need group control from ldap to web2py app's database feel free to set: If you need group control from ldap to web2py app's database feel free
to set:
auth.settings.login_methods.append(ldap_auth(...as usual..., auth.settings.login_methods.append(ldap_auth(...as usual...,
manage_groups=True, manage_groups=True,
@@ -114,7 +119,8 @@ def ldap_auth(server = 'ldap', port = None,
Where: Where:
manage_group - let web2py handle the groups from ldap manage_group - let web2py handle the groups from ldap
db - is the database object (need to have auth_user, auth_group, auth_membership) db - is the database object (need to have auth_user, auth_group,
auth_membership)
group_dn - the ldap branch of the groups group_dn - the ldap branch of the groups
group_name_attrib - the attribute where the group name is stored group_name_attrib - the attribute where the group name is stored
group_member_attrib - the attribute containing the group members name group_member_attrib - the attribute containing the group members name
@@ -134,10 +140,12 @@ def ldap_auth(server = 'ldap', port = None,
allowed_groups - a list with allowed ldap group names allowed_groups - a list with allowed ldap group names
group_dn - the ldap branch of the groups group_dn - the ldap branch of the groups
group_name_attrib - the attribute where the group name is stored group_name_attrib - the attribute where the group name is stored
group_member_attrib - the attibute containing the group members name group_member_attrib - the attribute containing the group members name
group_filterstr - as the filterstr but for group select group_filterstr - as the filterstr but for group select
If using Active Directory you must specify bind_dn and bind_pw for allowed_groups unless anonymous bind works. If using Active Directory you must specify bind_dn and bind_pw for
allowed_groups unless anonymous bind works.
You can set the logging level with the "logging_level" parameter, default You can set the logging level with the "logging_level" parameter, default
is "error" and can be set to error, warning, info, debug. is "error" and can be set to error, warning, info, debug.
""" """
@@ -150,6 +158,7 @@ def ldap_auth(server = 'ldap', port = None,
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
elif logging_level == 'debug': elif logging_level == 'debug':
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
def ldap_auth_aux(username, def ldap_auth_aux(username,
password, password,
ldap_server=server, ldap_server=server,
@@ -171,11 +180,12 @@ def ldap_auth(server = 'ldap', port = None,
manage_groups=manage_groups, manage_groups=manage_groups,
allowed_groups=allowed_groups, allowed_groups=allowed_groups,
db=db): db=db):
if password == '': if password == '': # http://tools.ietf.org/html/rfc4513#section-5.1.2
logger.warning('blank password not allowed') logger.warning('blank password not allowed')
return False return False
logger.debug('mode: [%s] manage_user: [%s] custom_scope: [%s] manage_groups: [%s]' % ( logger.debug('mode: [%s] manage_user: [%s] custom_scope: [%s]'
str(mode), str(manage_user), str(custom_scope), str(manage_groups))) ' manage_groups: [%s]' % (str(mode), str(manage_user),
str(custom_scope), str(manage_groups)))
if manage_user: if manage_user:
if user_firstname_attrib.count(':') > 0: if user_firstname_attrib.count(':') > 0:
(user_firstname_attrib, user_firstname_part) = user_firstname_attrib.split(':', 1) (user_firstname_attrib, user_firstname_part) = user_firstname_attrib.split(':', 1)
@@ -227,7 +237,8 @@ def ldap_auth(server = 'ldap', port = None,
filterstr), filterstr),
requested_attrs)[0][1] requested_attrs)[0][1]
if not isinstance(result, dict): if not isinstance(result, dict):
# result should be a dict in the form {'sAMAccountName': [username_bare]} # result should be a dict in the form
# {'sAMAccountName': [username_bare]}
logger.warning('User [%s] not found!' % username) logger.warning('User [%s] not found!' % username)
return False return False
if ldap_binddn: if ldap_binddn:
@@ -252,26 +263,22 @@ def ldap_auth(server = 'ldap', port = None,
dn = "cn=" + username + "," + ldap_basedn dn = "cn=" + username + "," + ldap_basedn
con.simple_bind_s(dn, password) con.simple_bind_s(dn, password)
if manage_user: if manage_user:
result = con.search_s( result = con.search_s(dn, ldap.SCOPE_BASE,
dn, ldap.SCOPE_BASE,
"(objectClass=*)", "(objectClass=*)",
[user_firstname_attrib, [user_firstname_attrib,
user_lastname_attrib, user_lastname_attrib,
user_mail_attrib] user_mail_attrib])[0][1]
)[0][1]
if ldap_mode == 'uid': if ldap_mode == 'uid':
# OpenLDAP (UID) # OpenLDAP (UID)
dn = "uid=" + username + "," + ldap_basedn dn = "uid=" + username + "," + ldap_basedn
con.simple_bind_s(dn, password) con.simple_bind_s(dn, password)
if manage_user: if manage_user:
result = con.search_s( result = con.search_s(dn, ldap.SCOPE_BASE,
dn, ldap.SCOPE_BASE,
"(objectClass=*)", "(objectClass=*)",
[user_firstname_attrib, [user_firstname_attrib,
user_lastname_attrib, user_lastname_attrib,
user_mail_attrib] user_mail_attrib])[0][1]
)[0][1]
if ldap_mode == 'company': if ldap_mode == 'company':
# no DNs or password needed to search directory # no DNs or password needed to search directory
@@ -280,8 +287,8 @@ def ldap_auth(server = 'ldap', port = None,
# bind anonymously # bind anonymously
con.simple_bind_s(dn, pw) con.simple_bind_s(dn, pw)
# search by e-mail address # search by e-mail address
filter = '(&(mail=' + ldap.filter.escape_filter_chars(username) + \ filter = '(&(mail=%s)(%s))' % (ldap.filter.escape_filter_chars(username),
')(' + filterstr + '))' filterstr)
# find the uid # find the uid
attrs = ['uid'] attrs = ['uid']
if manage_user: if manage_user:
@@ -299,43 +306,47 @@ def ldap_auth(server = 'ldap', port = None,
if ldap_mode == 'uid_r': if ldap_mode == 'uid_r':
# OpenLDAP (UID) with subtree search and multiple DNs # OpenLDAP (UID) with subtree search and multiple DNs
if type(ldap_basedn) == type([]): if isinstance(ldap_basedn, list):
basedns = ldap_basedn basedns = ldap_basedn
else: else:
basedns = [ldap_basedn] basedns = [ldap_basedn]
filter = '(&(uid=%s)(%s))' % (ldap.filter.escape_filter_chars(username), filterstr) filter = '(&(uid=%s)(%s))' % (ldap.filter.escape_filter_chars(username), filterstr)
finded = False found = False
for basedn in basedns: for basedn in basedns:
try: try:
result = con.search_s(basedn, ldap.SCOPE_SUBTREE, filter) result = con.search_s(basedn, ldap.SCOPE_SUBTREE,
filter)
if result: if result:
user_dn = result[0][0] user_dn = result[0][0]
# Check the password # Check the password
con.simple_bind_s(user_dn, password) con.simple_bind_s(user_dn, password)
finded = True found = True
break break
except ldap.LDAPError, detail: except ldap.LDAPError, detail:
(exc_type, exc_value) = sys.exc_info()[:2] (exc_type, exc_value) = sys.exc_info()[:2]
logger.warning("ldap_auth: searching %s for %s resulted in %s: %s\n" % logger.warning("ldap_auth: searching %s for %s resulted in %s: %s\n" %
(basedn, filter, exc_type, exc_value)) (basedn, filter, exc_type, exc_value))
if not finded: if not found:
logger.warning('User [%s] not found!' % username) logger.warning('User [%s] not found!' % username)
return False return False
result = result[0][1] result = result[0][1]
if ldap_mode == 'custom': if ldap_mode == 'custom':
# OpenLDAP (username_attrs) with subtree search and multiple DNs # OpenLDAP (username_attrs) with subtree search and
if type(ldap_basedn) == type([]): # multiple DNs
if isinstance(ldap_basedn, list):
basedns = ldap_basedn basedns = ldap_basedn
else: else:
basedns = [ldap_basedn] basedns = [ldap_basedn]
filter = '(&(%s=%s)(%s))' % (username_attrib, ldap.filter.escape_filter_chars(username), filterstr) filter = '(&(%s=%s)(%s))' % (username_attrib,
ldap.filter.escape_filter_chars(username),
filterstr)
if custom_scope == 'subtree': if custom_scope == 'subtree':
ldap_scope = ldap.SCOPE_SUBTREE ldap_scope = ldap.SCOPE_SUBTREE
elif custom_scope == 'base': elif custom_scope == 'base':
ldap_scope = ldap.SCOPE_BASE ldap_scope = ldap.SCOPE_BASE
elif custom_scope == 'onelevel': elif custom_scope == 'onelevel':
ldap_scope = ldap.SCOPE_ONELEVEL ldap_scope = ldap.SCOPE_ONELEVEL
finded = False found = False
for basedn in basedns: for basedn in basedns:
try: try:
result = con.search_s(basedn, ldap_scope, filter) result = con.search_s(basedn, ldap_scope, filter)
@@ -343,27 +354,27 @@ def ldap_auth(server = 'ldap', port = None,
user_dn = result[0][0] user_dn = result[0][0]
# Check the password # Check the password
con.simple_bind_s(user_dn, password) con.simple_bind_s(user_dn, password)
finded = True found = True
break break
except ldap.LDAPError, detail: except ldap.LDAPError, detail:
(exc_type, exc_value) = sys.exc_info()[:2] (exc_type, exc_value) = sys.exc_info()[:2]
logger.warning("ldap_auth: searching %s for %s resulted in %s: %s\n" % logger.warning("ldap_auth: searching %s for %s resulted in %s: %s\n" %
(basedn, filter, exc_type, exc_value)) (basedn, filter, exc_type, exc_value))
if not finded: if not found:
logger.warning('User [%s] not found!' % username) logger.warning('User [%s] not found!' % username)
return False return False
result = result[0][1] result = result[0][1]
if manage_user: if manage_user:
logger.info('[%s] Manage user data' % str(username)) logger.info('[%s] Manage user data' % str(username))
try: try:
if not user_firstname_part == None: if user_firstname_part is not None:
store_user_firstname = result[user_firstname_attrib][0].split(' ', 1)[user_firstname_part] store_user_firstname = result[user_firstname_attrib][0].split(' ', 1)[user_firstname_part]
else: else:
store_user_firstname = result[user_firstname_attrib][0] store_user_firstname = result[user_firstname_attrib][0]
except KeyError, e: except KeyError, e:
store_user_firstname = None store_user_firstname = None
try: try:
if not user_lastname_part == None: if user_lastname_part is not None:
store_user_lastname = result[user_lastname_attrib][0].split(' ', 1)[user_lastname_part] store_user_lastname = result[user_lastname_attrib][0].split(' ', 1)[user_lastname_part]
else: else:
store_user_lastname = result[user_lastname_attrib][0] store_user_lastname = result[user_lastname_attrib][0]
@@ -394,13 +405,11 @@ def ldap_auth(server = 'ldap', port = None,
user_in_db = db(db.auth_user.email == username) user_in_db = db(db.auth_user.email == username)
if user_in_db.count() > 0: if user_in_db.count() > 0:
user_in_db.update(first_name=store_user_firstname, user_in_db.update(first_name=store_user_firstname,
last_name = store_user_lastname, last_name=store_user_lastname)
)
else: else:
db.auth_user.insert(first_name=store_user_firstname, db.auth_user.insert(first_name=store_user_firstname,
last_name=store_user_lastname, last_name=store_user_lastname,
email = username email=username)
)
con.unbind() con.unbind()
if manage_groups: if manage_groups:
@@ -420,10 +429,10 @@ def ldap_auth(server = 'ldap', port = None,
def is_user_in_allowed_groups(username, def is_user_in_allowed_groups(username,
password=None, password=None,
allowed_groups = allowed_groups allowed_groups=allowed_groups):
):
""" """
Figure out if the username is a member of an allowed group in ldap or not Figure out if the username is a member of an allowed group
in ldap or not
""" """
# #
# Get all group name where the user is in actually in ldap # Get all group name where the user is in actually in ldap
@@ -442,8 +451,7 @@ def ldap_auth(server = 'ldap', port = None,
def do_manage_groups(username, def do_manage_groups(username,
password=None, password=None,
db = db, db=db):
):
""" """
Manage user groups Manage user groups
@@ -480,24 +488,27 @@ def ldap_auth(server = 'ldap', port = None,
if not db_user_id: if not db_user_id:
logging.error('There is no username or email for %s!' % username) logging.error('There is no username or email for %s!' % username)
raise raise
db_group_search = db((db.auth_membership.user_id == db_user_id) & \ db_group_search = db((db.auth_membership.user_id == db_user_id) &
(db.auth_user.id == db.auth_membership.user_id) & \ (db.auth_user.id == db.auth_membership.user_id) &
(db.auth_group.id == db.auth_membership.group_id)) (db.auth_group.id == db.auth_membership.group_id))
db_groups_of_the_user = list() db_groups_of_the_user = list()
db_group_id = dict() db_group_id = dict()
if db_group_search.count() > 0: if db_group_search.count() > 0:
for group in db_group_search.select(db.auth_group.id, db.auth_group.role, distinct = True): for group in db_group_search.select(db.auth_group.id,
db.auth_group.role,
distinct=True):
db_group_id[group.role] = group.id db_group_id[group.role] = group.id
db_groups_of_the_user.append(group.role) db_groups_of_the_user.append(group.role)
logging.debug('db groups of user %s: %s' % (username, str(db_groups_of_the_user))) logging.debug('db groups of user %s: %s' %
(username, str(db_groups_of_the_user)))
# #
# Delete user membership from groups where user is not anymore # Delete user membership from groups where user is not anymore
# ############################################################# # #############################################################
for group_to_del in db_groups_of_the_user: for group_to_del in db_groups_of_the_user:
if ldap_groups_of_the_user.count(group_to_del) == 0: if ldap_groups_of_the_user.count(group_to_del) == 0:
db((db.auth_membership.user_id == db_user_id) & \ db((db.auth_membership.user_id == db_user_id) &
(db.auth_membership.group_id == db_group_id[group_to_del])).delete() (db.auth_membership.group_id == db_group_id[group_to_del])).delete()
# #
@@ -513,25 +524,24 @@ def ldap_auth(server = 'ldap', port = None,
db.auth_membership.insert(user_id=db_user_id, db.auth_membership.insert(user_id=db_user_id,
group_id=gid) group_id=gid)
except: except:
logger.warning("[%s] Groups are not managed successully!" % str(username)) logger.warning("[%s] Groups are not managed successfully!" %
str(username))
import traceback import traceback
logger.debug(traceback.format_exc()) logger.debug(traceback.format_exc())
return False return False
return True return True
def init_ldap( def init_ldap(ldap_server=server,
ldap_server = server,
ldap_port=port, ldap_port=port,
ldap_basedn=base_dn, ldap_basedn=base_dn,
ldap_mode=mode, ldap_mode=mode,
secure=secure, secure=secure,
cert_path=cert_path, cert_path=cert_path,
cert_file = cert_file cert_file=cert_file):
):
""" """
Inicialize ldap connection Inicialize ldap connection
""" """
logger.info('[%s] Inicialize ldap connection' % str(ldap_server)) logger.info('[%s] Initialize ldap connection' % str(ldap_server))
if secure: if secure:
if not ldap_port: if not ldap_port:
ldap_port = 636 ldap_port = 636
@@ -557,8 +567,7 @@ def ldap_auth(server = 'ldap', port = None,
group_name_attrib=group_name_attrib, group_name_attrib=group_name_attrib,
group_member_attrib=group_member_attrib, group_member_attrib=group_member_attrib,
group_filterstr=group_filterstr, group_filterstr=group_filterstr,
ldap_mode = mode ldap_mode=mode):
):
""" """
Get all group names from ldap where the user is in Get all group names from ldap where the user is in
""" """
@@ -566,7 +575,7 @@ def ldap_auth(server = 'ldap', port = None,
# #
# Get all group name where the user is in actually in ldap # Get all group name where the user is in actually in ldap
# ######################################################### # #########################################################
# Inicialize ldap # Initialize ldap
if not group_dn: if not group_dn:
group_dn = base_dn group_dn = base_dn
con = init_ldap() con = init_ldap()
@@ -595,9 +604,9 @@ def ldap_auth(server = 'ldap', port = None,
con.simple_bind_s(username, password) con.simple_bind_s(username, password)
logger.debug('Ldap username connect...') logger.debug('Ldap username connect...')
# We have to use the full string # We have to use the full string
username = con.search_ext_s( username = con.search_ext_s(base_dn, ldap.SCOPE_SUBTREE,
base_dn, ldap.SCOPE_SUBTREE, "(&(sAMAccountName=%s)(%s))" %
"(&(sAMAccountName=%s)(%s))" % (ldap.filter.escape_filter_chars(username_bare), filterstr), ["cn"])[0][0] (ldap.filter.escape_filter_chars(username_bare), filterstr), ["cn"])[0][0]
else: else:
if ldap_binddn: if ldap_binddn:
# need to search directory with an bind_dn account 1st # need to search directory with an bind_dn account 1st
@@ -622,8 +631,6 @@ def ldap_auth(server = 'ldap', port = None,
logger.debug('User groups: %s' % ldap_groups_of_the_user) logger.debug('User groups: %s' % ldap_groups_of_the_user)
return list(ldap_groups_of_the_user) return list(ldap_groups_of_the_user)
if filterstr[0] == '(' and filterstr[-1] == ')': # rfc4515 syntax if filterstr[0] == '(' and filterstr[-1] == ')': # rfc4515 syntax
filterstr = filterstr[1:-1] # parens added again where used filterstr = filterstr[1:-1] # parens added again where used
return ldap_auth_aux return ldap_auth_aux