### settings.py ### # active directory authentication module AD_DNS_NAME = 'dc.com' # FQDN of your DC If using non-SSL use these #AD_LDAP_PORT=389 #AD_LDAP_URL='ldap://%s:%s' % (AD_DNS_NAME,AD_LDAP_PORT) # If using SSL use these: AD_LDAP_PORT=636 AD_LDAP_URL='ldaps://%s:%s' % (AD_DNS_NAME,AD_LDAP_PORT) AD_SEARCH_DN = 'ou=Staff,dc=example,dc=com' AD_NT4_DOMAIN = 'YOURDOMAIN' AD_SEARCH_FIELDS = ['mail','givenName','sn','sAMAccountName','memberOf'] AD_MEMBERSHIP_ADMIN = ['admingroup'] # this ad group gets superuser status in django AD_MEMBERSHIP_REQ = AD_MEMBERSHIP_ADMIN + ['anothergroup', 'secondarygroup'] # only members of this group can access AD_CERT_FILE = '/path/to/certile.pem' # this is the certificate of the Certificate Authority issuing your DCs certificate AD_DEBUG=True AD_DEBUG_FILE='/tmp/ldap.debug' AUTHENTICATION_BACKENDS = ( '... .backend.ActiveDirectoryAuthenticationBackend', 'django.contrib.auth.backends.ModelBackend' ) ### backend.py ### import ldap import os import re import datetime from django.conf import settings from django.contrib.auth.models import User class ActiveDirectoryAuthenticationBackend: """ This authentication backend authenticates against Active directory. It updates the user objects according to the settings in the AD and is able to map specific groups to give users admin rights. """ def __init__(self): """ initialise a debuging if enabled """ self.debug = settings.AD_DEBUG if len(settings.AD_DEBUG_FILE) > 0 and self.debug: self.debugFile = settings.AD_DEBUG_FILE # is the debug file accessible? if not os.path.exists(self.debugFile): open(self.debugFile,'w').close() elif not os.access(self.debugFile, os.W_OK): raise IOError("Debug File is not writable") else: self.debugFile = None def authenticate(self,username=None,password=None): try: if len(password) == 0: return None ldap.set_option(ldap.OPT_X_TLS_CACERTFILE,settings.AD_CERT_FILE) l = ldap.initialize(settings.AD_LDAP_URL) l.set_option(ldap.OPT_PROTOCOL_VERSION, 3) binddn = "%s@%s" % (username,settings.AD_NT4_DOMAIN) l.simple_bind_s(binddn,password) l.unbind_s() return self.get_or_create_user(username,password) except ImportError: self.debug_write('import error in authenticate') except ldap.INVALID_CREDENTIALS: self.debug_write('%s: Invalid Credentials' % username) def get_or_create_user(self, username, password): """ create or update the User object """ # get user info from AD userInfo = self.get_user_info(username, password) # is there already a user? try: user = User.objects.get(username=username) except User.DoesNotExist: if userInfo is not None: user = User(username=username, password=password) self.debug_write('create new user %s' % username) ## update the user objects # group mapping for the access rights if userInfo['isAdmin']: user.is_staff = True user.is_superuser = True elif userInfo: user.is_staff = False user.is_superuser = False else: user.is_active = False # personal data user.first_name = userInfo['first_name'] user.last_name = userInfo['last_name'] user.mail = userInfo['mail'] # cache the AD password user.set_password(password) user.save() return user def get_user(self, user_id): try: return User.objects.get(pk=user_id) except User.DoesNotExist: return None def debug_write(self,message): """ handle debug messages """ if self.debugFile is not None: fObj = open(self.debugFile, 'a') now = datetime.datetime.now().strftime('%Y-%m-%d %H:%M') fObj.write("%s\t%s\n" % (now,message)) fObj.close() def check_group(self,membership): """ evaluate ADS group memberships """ self.debug_write("Evaluating group membership") isValid = False isAdmin = False pattern = re.compile(r'^CN=(?P[\w|\d]+),') for group in membership: groupMatch = pattern.match(group) if groupMatch: thisGroup = groupMatch.group('groupName') if thisGroup in settings.AD_MEMBERSHIP_REQ: isValid = True if thisGroup in settings.AD_MEMBERSHIP_ADMIN: isAdmin = True if isAdmin: self.debug_write('is admin user') elif isValid: self.debug_write('is normal user') else: self.debug_write('does not have the AD group membership needed') return isAdmin, isValid def get_user_info(self, username, password): """ get user info from ADS to a dictionary """ try: userInfo = { 'username' : username, 'password' : password, } # prepare LDAP connection ldap.set_option(ldap.OPT_X_TLS_CACERTFILE,settings.AD_CERT_FILE) ldap.set_option(ldap.OPT_REFERRALS,0) # DO NOT TURN THIS OFF OR SEARCH WON'T WORK! # initialize self.debug_write('ldap.initialize...') l = ldap.initialize(settings.AD_LDAP_URL) l.set_option(ldap.OPT_PROTOCOL_VERSION, 3) # bind binddn = "%s@%s" % (username,settings.AD_NT4_DOMAIN) self.debug_write('ldap.bind %s' % binddn) l.bind_s(binddn,password) # search self.debug_write('search...') result = l.search_ext_s(settings.AD_SEARCH_DN,ldap.SCOPE_SUBTREE,"sAMAccountName=%s" % username,settings.AD_SEARCH_FIELDS)[0][1] self.debug_write("results in %s" % result) # Validate that they are a member of review board group if result.has_key('memberOf'): membership = result['memberOf'] else: self.debug_write('AD user has no group memberships') return None # user ADS Groups isAdmin, isValid = self.check_group(membership) if not isValid: return None userInfo['isAdmin'] = isAdmin # get user info from ADS # get email if result.has_key('mail'): mail = result['mail'][0] else: mail = "" userInfo['mail'] = mail self.debug_write("mail=%s" % mail) # get surname if result.has_key('sn'): last_name = result['sn'][0] else: last_name = None userInfo['last_name'] = last_name self.debug_write("sn=%s" % last_name) # get display name if result.has_key('givenName'): first_name = result['givenName'][0] else: first_name = None userInfo['first_name'] = first_name self.debug_write("first_name=%s" % first_name) # LDAP unbind l.unbind_s() return userInfo except Exception, e: self.debug_write("exception caught!") self.debug_write(e) return None