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

new ldap_auth.py has manage_groups, thanks Gyuris

Browse Source
This commit is contained in:
Massimo Di Pierro
2012-03-01 08:55:02 -06:00
parent 02e7f38da7
commit 59768bede4
2 changed files with 189 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
Version 1.99.4 (2012-02-29 13:38:37) stable Version 1.99.4 (2012-03-01 08:54:35) stable
gluon/contrib/login_methods/ldap_auth.py
+188 -49
View File
@@ -2,15 +2,22 @@ import sys
import logging import logging
try: try:
import ldap import ldap
ldap.set_option(ldap.OPT_REFERRALS, 0) import ldap.filter
ldap.set_option( ldap.OPT_REFERRALS, 0 )
except Exception, e: 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,
mode='uid', secure=False, cert_path=None, bind_dn=None, bind_pw=None, filterstr='objectClass=*'): bind_dn = None, bind_pw = None, filterstr = 'objectClass=*',
manage_groups = False,
db = None,
group_dn = 'ou=groups,dc=domain,dc=com',
group_name_attrib = 'cn',
group_member_attrib = 'memberUid',
group_filterstr = 'objectClass=*' ):
""" """
to use ldap login with MS Active Directory:: to use ldap login with MS Active Directory::
@@ -49,128 +56,260 @@ 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 If you need to restrict the set of allowed users (e.g. to members of a department) then specify
a rfc4515 search filter string. a rfc4515 search filter string.
- currently only implemented for mode in ['ad', 'company', 'uid_r'] - currently only implemented for mode in ['ad', 'company', 'uid_r']
If you need group controll from ldap to web2py app's database feel free to set:
auth.settings.login_methods.append(ldap_auth(...as usual...,
manage_groups = True,
db = db,
group_dn = 'ou=Groups,dc=domain,dc=com',
group_name_attrib = 'cn',
group_member_attrib = 'memberUid',
group_filterstr = 'objectClass=*'
))
Where:
manage_group - let web2py handle the groups from ldap
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_filterstr - as the filterstr but for group select
""" """
def ldap_auth_aux(username, def ldap_auth_aux( username,
password, password,
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,
ldap_binddn=bind_dn, ldap_binddn = bind_dn,
ldap_bindpw=bind_pw, ldap_bindpw = bind_pw,
secure=secure, secure = secure,
cert_path=cert_path, cert_path = cert_path,
filterstr=filterstr): filterstr = filterstr,
manage_groups = manage_groups ):
try: try:
if secure: if secure:
if not ldap_port: if not ldap_port:
ldap_port = 636 ldap_port = 636
con = ldap.initialize( con = ldap.initialize(
"ldaps://" + ldap_server + ":" + str(ldap_port)) "ldaps://" + ldap_server + ":" + str( ldap_port ) )
if cert_path: if cert_path:
con.set_option(ldap.OPT_X_TLS_CACERTDIR, cert_path) con.set_option( ldap.OPT_X_TLS_CACERTDIR, cert_path )
else: else:
if not ldap_port: if not ldap_port:
ldap_port = 389 ldap_port = 389
con = ldap.initialize( con = ldap.initialize(
"ldap://" + ldap_server + ":" + str(ldap_port)) "ldap://" + ldap_server + ":" + str( ldap_port ) )
if ldap_mode == 'ad': if ldap_mode == 'ad':
# Microsoft Active Directory # Microsoft Active Directory
if '@' not in username: if '@' not in username:
domain = [] domain = []
for x in ldap_basedn.split(','): for x in ldap_basedn.split( ',' ):
if "DC=" in x.upper(): if "DC=" in x.upper():
domain.append(x.split('=')[-1]) domain.append( x.split( '=' )[-1] )
username = "%s@%s" % (username, '.'.join(domain)) username = "%s@%s" % ( username, '.'.join( domain ) )
username_bare = username.split("@")[0] username_bare = username.split( "@" )[0]
con.set_option(ldap.OPT_PROTOCOL_VERSION, 3) con.set_option( ldap.OPT_PROTOCOL_VERSION, 3 )
# In cases where ForestDnsZones and DomainDnsZones are found, # In cases where ForestDnsZones and DomainDnsZones are found,
# result will look like the following: # result will look like the following:
# ['ldap://ForestDnsZones.domain.com/DC=ForestDnsZones,DC=domain,DC=com'] # ['ldap://ForestDnsZones.domain.com/DC=ForestDnsZones,DC=domain,DC=com']
if ldap_binddn: if ldap_binddn:
# need to search directory with an admin account 1st # need to search directory with an admin account 1st
con.simple_bind_s(ldap_binddn, ldap_bindpw) con.simple_bind_s( ldap_binddn, ldap_bindpw )
else: else:
# credentials should be in the form of username@domain.tld # credentials should be in the form of username@domain.tld
con.simple_bind_s(username, password) con.simple_bind_s( username, password )
# this will throw an index error if the account is not found # this will throw an index error if the account is not found
# in the ldap_basedn # in the ldap_basedn
result = con.search_ext_s( result = con.search_ext_s(
ldap_basedn, ldap.SCOPE_SUBTREE, ldap_basedn, ldap.SCOPE_SUBTREE,
"(&(sAMAccountName=%s)(%s))" % (username_bare, filterstr), ["sAMAccountName"])[0][1] "(&(sAMAccountName=%s)(%s))" % ( ldap.filter.escape_filter_chars( username_bare ), filterstr ), ["sAMAccountName"] )[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]}
return False return False
if ldap_binddn: if ldap_binddn:
# We know the user exists & is in the correct OU # We know the user exists & is in the correct OU
# so now we just check the password # so now we just check the password
con.simple_bind_s(username, password) con.simple_bind_s( username, password )
if ldap_mode == 'domino': if ldap_mode == 'domino':
# Notes Domino # Notes Domino
if "@" in username: if "@" in username:
username = username.split("@")[0] username = username.split( "@" )[0]
con.simple_bind_s(username, password) con.simple_bind_s( username, password )
if ldap_mode == 'cn': if ldap_mode == 'cn':
# OpenLDAP (CN) # OpenLDAP (CN)
dn = "cn=" + username + "," + ldap_basedn dn = "cn=" + username + "," + ldap_basedn
con.simple_bind_s(dn, password) con.simple_bind_s( dn, password )
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 ldap_mode == 'company': if ldap_mode == 'company':
# no DNs or password needed to search directory # no DNs or password needed to search directory
dn = "" dn = ""
pw = "" pw = ""
# 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=' + username + ')(' + filterstr + '))' filter = '(&(mail=' + ldap.filter.escape_filter_chars( username ) + \
')(' + filterstr + '))'
# find the uid # find the uid
attrs = ['uid'] attrs = ['uid']
# perform the actual search # perform the actual search
company_search_result=con.search_s(ldap_basedn, company_search_result = con.search_s( ldap_basedn,
ldap.SCOPE_SUBTREE, ldap.SCOPE_SUBTREE,
filter, attrs) filter, attrs )
dn = company_search_result[0][0] dn = company_search_result[0][0]
# perform the real authentication test # perform the real authentication test
con.simple_bind_s(dn, password) con.simple_bind_s( dn, password )
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 type( ldap_basedn ) == type( [] ):
basedns = ldap_basedn basedns = ldap_basedn
else: else:
basedns = [ldap_basedn] basedns = [ldap_basedn]
filter = '(&(uid=%s)(%s))' % (username, filterstr) filter = '(&(uid=%s)(%s))' % ( ldap.filter.escape_filter_chars( username ), filterstr )
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 )
con.unbind() con.unbind()
if manage_groups:
do_manage_groups( username )
return True return True
except ldap.LDAPError, detail: except ldap.LDAPError, detail:
(exc_type, exc_value) = sys.exc_info()[:2] ( exc_type, exc_value ) = sys.exc_info()[:2]
sys.stderr.write("ldap_auth: searching %s for %s resulted in %s: %s\n" % sys.stderr.write( "ldap_auth: searching %s for %s resulted in %s: %s\n" %
(basedn, filter, exc_type, exc_value)) ( basedn, filter, exc_type, exc_value ) )
return False return False
con.unbind() con.unbind()
if manage_groups:
do_manage_groups( username )
return True return True
except ldap.LDAPError, e: except ldap.LDAPError, e:
import traceback
logging.error( traceback.format_exc() )
return False return False
except IndexError, ex: # for AD membership test except IndexError, ex: # for AD membership test
return False return False
def do_manage_groups( username,
ldap_server = server,
ldap_port = port,
ldap_basedn = base_dn,
ldap_mode = mode,
ldap_binddn = bind_dn,
ldap_bindpw = bind_pw,
secure = secure,
cert_path = cert_path,
filterstr = filterstr,
group_dn = group_dn,
group_name_attrib = group_name_attrib,
group_member_attrib = group_member_attrib,
group_filterstr = group_filterstr,
db = db,
):
'''
Manage user groups
Get all user's group from ldap and refresh the already stored
ones in web2py's application database or create new groups
according to ldap.
'''
#
# Get all group name where the user is in actually in ldap
# #########################################################
# Inicialize ldap
if secure:
if not ldap_port:
ldap_port = 636
con = ldap.initialize(
"ldaps://" + ldap_server + ":" + str( ldap_port ) )
if cert_path:
con.set_option( ldap.OPT_X_TLS_CACERTDIR, cert_path )
else:
if not ldap_port:
ldap_port = 389
con = ldap.initialize(
"ldap://" + ldap_server + ":" + str( ldap_port ) )
if ldap_binddn:
# need to search directory with an bind_dn account 1st
con.simple_bind_s( ldap_binddn, ldap_bindpw )
else:
# bind as anonymous
con.simple_bind_s( '', '' )
# search for groups where user is in
filter = '(&(%s=%s)(%s))' % ( ldap.filter.escape_filter_chars( group_member_attrib ),
ldap.filter.escape_filter_chars( username ),
group_filterstr )
group_search_result = con.search_s( group_dn,
ldap.SCOPE_SUBTREE,
filter, [group_name_attrib] )
ldap_groups_of_the_user = list()
for group_row in group_search_result:
group = group_row[1]
ldap_groups_of_the_user.extend( group[group_name_attrib] )
#
# Get all group name where the user is in actually in local db
# #############################################################
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
if not db_user_id:
logging.error( 'There is no username or email for %s!' % username )
raise
db_group_search = db( ( db.auth_membership.user_id == db_user_id ) & \
( db.auth_user.id == db.auth_membership.user_id ) & \
( db.auth_group.id == db.auth_membership.group_id ) )
db_groups_of_the_user = list()
db_group_id = dict()
if db_group_search.count() > 0:
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_groups_of_the_user.append( group.role )
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
# #############################################################
for group_to_del in db_groups_of_the_user:
if ldap_groups_of_the_user.count( group_to_del ) == 0:
db( ( db.auth_membership.user_id == db_user_id ) & \
( db.auth_membership.group_id == db_group_id[group_to_del] ) ).delete()
#
# Create user membership in groups where user is not in already
# ##############################################################
for group_to_add in ldap_groups_of_the_user:
if db_groups_of_the_user.count( group_to_add ) == 0:
if db( db.auth_group.role == group_to_add ).count() == 0:
gid = db.auth_group.insert( role = group_to_add,
description = 'Generated from LDAP' )
else:
gid = db( db.auth_group.role == group_to_add ).select( db.auth_group.id ).first().id
db.auth_membership.insert( user_id = db_user_id,
group_id = gid )
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