ldap certificate file support, thanks Gyuris
This commit is contained in:
@@ -1 +1 @@
|
||||
Version 1.99.7 (2012-03-11 19:38:42) dev
|
||||
Version 1.99.7 (2012-03-12 10:41:17) dev
|
||||
|
||||
@@ -10,15 +10,20 @@ except Exception, e:
|
||||
|
||||
def ldap_auth( server = 'ldap', port = None,
|
||||
base_dn = 'ou=users,dc=domain,dc=com',
|
||||
mode = 'uid', secure = False, cert_path = None,
|
||||
mode = 'uid', secure = False, cert_path = None, cert_file = None,
|
||||
bind_dn = None, bind_pw = None, filterstr = 'objectClass=*',
|
||||
allowed_groups = None,
|
||||
manage_user = False,
|
||||
user_firstname_attrib = 'cn:1',
|
||||
user_lastname_attrib = 'cn:2',
|
||||
user_mail_attrib = 'mail',
|
||||
manage_groups = False,
|
||||
db = None,
|
||||
group_dn = 'ou=groups,dc=domain,dc=com',
|
||||
group_dn = None,
|
||||
group_name_attrib = 'cn',
|
||||
group_member_attrib = 'memberUid',
|
||||
group_filterstr = 'objectClass=*' ):
|
||||
|
||||
"""
|
||||
to use ldap login with MS Active Directory::
|
||||
|
||||
@@ -50,6 +55,8 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
base_dn='ou=Users,dc=domain,dc=com'))
|
||||
|
||||
If using secure ldaps:// pass secure=True and cert_path="..."
|
||||
If ldap is using GnuTLS then you need cert_file="..." instead cert_path 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.
|
||||
- currently only implemented for Active Directory
|
||||
@@ -57,6 +64,25 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
If you need to restrict the set of allowed users (e.g. to members of a department) then specify
|
||||
a rfc4515 search filter string.
|
||||
- currently only implemented for mode in ['ad', 'company', 'uid_r']
|
||||
|
||||
You can manage user attribute first name, last name, email from ldap:
|
||||
auth.settings.login_methods.append(ldap_auth(...as usual...,
|
||||
manage_user = True,
|
||||
user_firstname_attrib = 'cn:1',
|
||||
user_lastname_attrib = 'cn:2',
|
||||
user_mail_attrib = 'mail'
|
||||
))
|
||||
|
||||
Where:
|
||||
manage_user - let web2py handle user data from ldap
|
||||
user_firstname_attrib - the attribute containing the user's first name
|
||||
optionally you can specify parts.
|
||||
Example: cn: "John Smith" - 'cn:1' = 'John'
|
||||
user_lastname_attrib - the attribute containing the user's last name
|
||||
optionally you can specify parts.
|
||||
Example: cn: "John Smith" - 'cn:2' = 'Smith'
|
||||
user_mail_attrib - the attribure containing the user's email address
|
||||
|
||||
|
||||
If you need group control from ldap to web2py app's database feel free to set:
|
||||
|
||||
@@ -74,7 +100,7 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
db - is the database object (need to have auth_user, auth_group, auth_membership)
|
||||
group_dn - the ldap branch of the groups
|
||||
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
|
||||
|
||||
You can restrict login access to specific groups if you specify:
|
||||
@@ -105,9 +131,29 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
ldap_bindpw = bind_pw,
|
||||
secure = secure,
|
||||
cert_path = cert_path,
|
||||
cert_file = cert_file,
|
||||
filterstr = filterstr,
|
||||
manage_user = manage_user,
|
||||
user_firstname_attrib = user_firstname_attrib,
|
||||
user_lastname_attrib = user_lastname_attrib,
|
||||
user_mail_attrib = user_mail_attrib,
|
||||
manage_groups = manage_groups,
|
||||
allowed_groups = allowed_groups ):
|
||||
allowed_groups = allowed_groups,
|
||||
db = db ):
|
||||
if manage_user:
|
||||
if user_firstname_attrib.count( ':' ) > 0:
|
||||
( user_firstname_attrib, user_firstname_part ) = user_firstname_attrib.split( ':', 1 )
|
||||
user_firstname_part = ( int( user_firstname_part ) - 1 )
|
||||
else:
|
||||
user_firstname_part = None
|
||||
if user_lastname_attrib.count( ':' ) > 0:
|
||||
( user_lastname_attrib, user_lastname_part ) = user_lastname_attrib.split( ':', 1 )
|
||||
user_lastname_part = ( int( user_lastname_part ) - 1 )
|
||||
else:
|
||||
user_lastname_part = None
|
||||
user_firstname_attrib = ldap.filter.escape_filter_chars( user_firstname_attrib )
|
||||
user_lastname_attrib = ldap.filter.escape_filter_chars( user_lastname_attrib )
|
||||
user_mail_attrib = ldap.filter.escape_filter_chars( user_mail_attrib )
|
||||
try:
|
||||
con = init_ldap()
|
||||
if allowed_groups:
|
||||
@@ -134,9 +180,16 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
con.simple_bind_s( username, password )
|
||||
# this will throw an index error if the account is not found
|
||||
# in the ldap_basedn
|
||||
requested_attrs = ['sAMAccountName']
|
||||
if manage_user:
|
||||
requested_attrs.extend( [user_firstname_attrib,
|
||||
user_lastname_attrib,
|
||||
user_mail_attrib] )
|
||||
result = con.search_ext_s(
|
||||
ldap_basedn, ldap.SCOPE_SUBTREE,
|
||||
"(&(sAMAccountName=%s)(%s))" % ( ldap.filter.escape_filter_chars( username_bare ), filterstr ), ["sAMAccountName"] )[0][1]
|
||||
"(&(sAMAccountName=%s)(%s))" % ( ldap.filter.escape_filter_chars( username_bare ),
|
||||
filterstr ),
|
||||
requested_attrs )[0][1]
|
||||
if not isinstance( result, dict ):
|
||||
# result should be a dict in the form {'sAMAccountName': [username_bare]}
|
||||
return False
|
||||
@@ -150,16 +203,37 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
if "@" in username:
|
||||
username = username.split( "@" )[0]
|
||||
con.simple_bind_s( username, password )
|
||||
if manage_user:
|
||||
# TODO: sorry I have no clue how to query attrs in domino
|
||||
result = {user_firstname_attrib: username,
|
||||
user_lastname_attrib: None,
|
||||
user_mail_attrib: None}
|
||||
|
||||
if ldap_mode == 'cn':
|
||||
# OpenLDAP (CN)
|
||||
dn = "cn=" + username + "," + ldap_basedn
|
||||
con.simple_bind_s( dn, password )
|
||||
if manage_user:
|
||||
result = con.search_s(
|
||||
dn, ldap.SCOPE_BASE,
|
||||
"(objectClass=*)",
|
||||
[user_firstname_attrib,
|
||||
user_lastname_attrib,
|
||||
user_mail_attrib]
|
||||
)[0][1]
|
||||
|
||||
if ldap_mode == 'uid':
|
||||
# OpenLDAP (UID)
|
||||
dn = "uid=" + username + "," + ldap_basedn
|
||||
con.simple_bind_s( dn, password )
|
||||
if manage_user:
|
||||
result = con.search_s(
|
||||
dn, ldap.SCOPE_BASE,
|
||||
"(objectClass=*)",
|
||||
[user_firstname_attrib,
|
||||
user_lastname_attrib,
|
||||
user_mail_attrib]
|
||||
)[0][1]
|
||||
|
||||
if ldap_mode == 'company':
|
||||
# no DNs or password needed to search directory
|
||||
@@ -172,11 +246,16 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
')(' + filterstr + '))'
|
||||
# find the uid
|
||||
attrs = ['uid']
|
||||
if manage_user:
|
||||
attrs.extend( [user_firstname_attrib,
|
||||
user_lastname_attrib,
|
||||
user_mail_attrib] )
|
||||
# perform the actual search
|
||||
company_search_result = con.search_s( ldap_basedn,
|
||||
ldap.SCOPE_SUBTREE,
|
||||
filter, attrs )
|
||||
dn = company_search_result[0][0]
|
||||
result = company_search_result[0][1]
|
||||
# perform the real authentication test
|
||||
con.simple_bind_s( dn, password )
|
||||
|
||||
@@ -187,6 +266,7 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
else:
|
||||
basedns = [ldap_basedn]
|
||||
filter = '(&(uid=%s)(%s))' % ( ldap.filter.escape_filter_chars( username ), filterstr )
|
||||
finded = False
|
||||
for basedn in basedns:
|
||||
try:
|
||||
result = con.search_s( basedn, ldap.SCOPE_SUBTREE, filter )
|
||||
@@ -194,17 +274,64 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
user_dn = result[0][0]
|
||||
# Check the password
|
||||
con.simple_bind_s( user_dn, password )
|
||||
con.unbind()
|
||||
if manage_groups:
|
||||
do_manage_groups( username )
|
||||
return True
|
||||
finded = True
|
||||
break
|
||||
except ldap.LDAPError, detail:
|
||||
( exc_type, exc_value ) = sys.exc_info()[:2]
|
||||
sys.stderr.write( "ldap_auth: searching %s for %s resulted in %s: %s\n" %
|
||||
( basedn, filter, exc_type, exc_value ) )
|
||||
return False
|
||||
|
||||
if not finded:
|
||||
return False
|
||||
result = result[0][1]
|
||||
if manage_user:
|
||||
try:
|
||||
if not user_firstname_part == None:
|
||||
store_user_firstname = result[user_firstname_attrib][0].split( ' ', 1 )[user_firstname_part]
|
||||
else:
|
||||
store_user_firstname = result[user_firstname_attrib][0]
|
||||
except KeyError, e:
|
||||
store_user_firstname = None
|
||||
try:
|
||||
if not user_lastname_part == None:
|
||||
store_user_lastname = result[user_lastname_attrib][0].split( ' ', 1 )[user_lastname_part]
|
||||
else:
|
||||
store_user_lastname = result[user_lastname_attrib][0]
|
||||
except KeyError, e:
|
||||
store_user_lastname = None
|
||||
try:
|
||||
store_user_mail = result[user_mail_attrib][0]
|
||||
except KeyError, e:
|
||||
store_user_mail = None
|
||||
try:
|
||||
#
|
||||
# user as username
|
||||
# #################
|
||||
user_in_db = db( db.auth_user.username == username )
|
||||
if user_in_db.count() > 0:
|
||||
user_in_db.update( first_name = store_user_firstname,
|
||||
last_name = store_user_lastname,
|
||||
email = store_user_mail )
|
||||
else:
|
||||
db.auth_user.insert( first_name = store_user_firstname,
|
||||
last_name = store_user_lastname,
|
||||
email = store_user_mail,
|
||||
username = username )
|
||||
except:
|
||||
#
|
||||
# user as email
|
||||
# ##############
|
||||
user_in_db = db( db.auth_user.email == username )
|
||||
if user_in_db.count() > 0:
|
||||
user_in_db.update( first_name = store_user_firstname,
|
||||
last_name = store_user_lastname,
|
||||
)
|
||||
else:
|
||||
db.auth_user.insert( first_name = store_user_firstname,
|
||||
last_name = store_user_lastname,
|
||||
email = username
|
||||
)
|
||||
con.unbind()
|
||||
|
||||
if manage_groups:
|
||||
do_manage_groups( username )
|
||||
|
||||
@@ -256,7 +383,19 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
try:
|
||||
db_user_id = db( db.auth_user.username == username ).select( db.auth_user.id ).first().id
|
||||
except:
|
||||
db_user_id = db( db.auth_user.email == username ).select( db.auth_user.id ).first().id
|
||||
try:
|
||||
db_user_id = db( db.auth_user.email == username ).select( db.auth_user.id ).first().id
|
||||
except AttributeError, e:
|
||||
#
|
||||
# There is no user in local db
|
||||
# We create one
|
||||
# ##############################
|
||||
try:
|
||||
db_user_id = db.auth_user.insert( username = username,
|
||||
first_name = username )
|
||||
except AttributeError, e:
|
||||
db_user_id = db.auth_user.insert( email = username,
|
||||
first_name = username )
|
||||
if not db_user_id:
|
||||
logging.error( 'There is no username or email for %s!' % username )
|
||||
raise
|
||||
@@ -300,6 +439,7 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
ldap_mode = mode,
|
||||
secure = secure,
|
||||
cert_path = cert_path,
|
||||
cert_file = cert_file
|
||||
):
|
||||
'''
|
||||
Inicialize ldap connection
|
||||
@@ -311,6 +451,8 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
"ldaps://" + ldap_server + ":" + str( ldap_port ) )
|
||||
if cert_path:
|
||||
con.set_option( ldap.OPT_X_TLS_CACERTDIR, cert_path )
|
||||
if cert_file:
|
||||
con.set_option( ldap.OPT_X_TLS_CACERTFILE, cert_file )
|
||||
else:
|
||||
if not ldap_port:
|
||||
ldap_port = 389
|
||||
@@ -319,6 +461,7 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
return con
|
||||
|
||||
def get_user_groups_from_ldap( username,
|
||||
base_dn = base_dn,
|
||||
ldap_binddn = bind_dn,
|
||||
ldap_bindpw = bind_pw,
|
||||
group_dn = group_dn,
|
||||
@@ -333,6 +476,8 @@ def ldap_auth( server = 'ldap', port = None,
|
||||
# Get all group name where the user is in actually in ldap
|
||||
# #########################################################
|
||||
# Inicialize ldap
|
||||
if not group_dn:
|
||||
group_dn = base_dn
|
||||
con = init_ldap()
|
||||
if ldap_binddn:
|
||||
# need to search directory with an bind_dn account 1st
|
||||
|
||||
Reference in New Issue
Block a user