Python + Active Directory + Linux

So, this is really pretty old, but I wanted to share it, since at the time, it took me a while to gather a lot of this information: Managing Active Directory (LDAP) via Linux + Python. There have probably been other posts on this since, but I wanted to put it out there.

I had managed OpenLDAP previously, and then we migrated to Active Directory (Windows Server 2008 at the time). There were a few gotchas that I wasn't expecting.

First, is the default 1,000 entry search results returned by AD. You can change this limit (setting) in Active Directory, but when you think about it, it makes sense to keep it -- if you have a huge directory and are running lots of searches that have huge results, this could definitely thrash your domain controllers.

For your programs to get around this, you need to do use "paged results" (otherwise you'll get an error stating the max returned entries is 1,000).

We are a RHEL shop, but at the time when I worked on this, the default python-ldap package didn't include paged results support (can't remember the version). So, on RHEL / CentOS, grab the latest python-ldap package, and do something like this:

yum remove python-ldap
yum groupinstall "Development Tools"
yum install python-devel
yum install openldap-devel
yum install openssl-devel
tar xvfz python-ldap-2.3.8.tar.gz
cd python-ldap-2.3.8
python setup.py build
python setup.py bdist_rpm
rpm -ivh dist/python-ldap-2.3.8-0.x86_64.rpm

Here is an example in Python of one of those paged searches. I'm pretty sure I grabbed the original from another post somewhere else on the 'net, but I wanted to share it for completeness:

import ldap
from ldap.controls import SimplePagedResultsControl
import sys
import ldap.modlist as modlist

LDAP_SERVER = "ldaps://dc.host.com"
BIND_DN = "Operator@host.com"
BIND_PASS = "password"
USER_FILTER = "(&(objectClass=person)(primaryGroupID=7235))"
USER_BASE = "ou=Special Peeps,ou=My Users,dc=host,dc=com"
PAGE_SIZE = 10

# LDAP connection
try:
  ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, 0)
  ldap_connection = ldap.initialize(LDAP_SERVER)
  ldap_connection.simple_bind_s(BIND_DN, BIND_PASS)
except ldap.LDAPError, e:
  sys.stderr.write('Error connecting to LDAP server: ' + str(e) + '\n')
  sys.exit(1)

# Lookup usernames from LDAP via paged search
paged_results_control = SimplePagedResultsControl(
  ldap.LDAP_CONTROL_PAGE_OID, True, (PAGE_SIZE, ''))
accounts = []
pages = 0
while True:
  serverctrls = [paged_results_control]
  try:
      msgid = ldap_connection.search_ext(USER_BASE,
                                         ldap.SCOPE_ONELEVEL,
                                         USER_FILTER,
                                         attrlist=['employeeID',
                                                   'sAMAccountName'],
                                         serverctrls=serverctrls)
  except ldap.LDAPError, e:
      sys.stderr.write('Error performing user paged search: ' +
                       str(e) + '\n')
      sys.exit(1)
  try:
      unused_code, results, unused_msgid, serverctrls = \
                 ldap_connection.result3(msgid)
  except ldap.LDAPError, e:
      sys.stderr.write('Error getting user paged search results: ' +
                       str(e) + '\n')
      sys.exit(1)
  for result in results:
      pages += 1
      accounts.append(result)
  cookie = None
  for serverctrl in serverctrls:
      if serverctrl.controlType == ldap.LDAP_CONTROL_PAGE_OID:
          unused_est, cookie = serverctrl.controlValue
          if cookie:
              paged_results_control.controlValue = (PAGE_SIZE, cookie)
          break
  if not cookie:
      break

# LDAP unbind
ldap_connection.unbind_s()

# Make dictionary with user data
user_map = {}
for entry in accounts:
  if entry[1].has_key('employeeID') and \
     entry[1].has_key('sAMAccountName'):
      user_map[entry[1]['employeeID'][0]] = entry[1]['sAMAccountName'][0]

In the above block, I included an example of the connection setup, and retrieving attributes from the result set (not really specific to AD, but someone might find it helpful). Below are some more Python-AD examples, but they are just little snippets (not necessarily complete) of the action.

Changing an Active Directory user's password:

PASSWORD_ATTR = "unicodePwd"
user_dn = user_results[0][1]['distinguishedName'][0]
username = sys.argv[1]
password = getpass.getpass("New password: ")

# Set AD password
unicode_pass = unicode("\"" + password + "\"", "iso-8859-1")
password_value = unicode_pass.encode("utf-16-le")
add_pass = [(ldap.MOD_REPLACE, PASSWORD_ATTR, [password_value])]

# Replace password
try:
  ldap_connection.modify_s(user_dn, add_pass)
  print "Active Directory password for", username, \
        "was set successfully!"
except ldap.LDAPError, e:
  sys.stderr.write('Error setting AD password for: ' + username + '\n')
  sys.stderr.write('Message: ' + str(e) + '\n')
  sys.exit(1)

Create an Active Directory user account:

def CreateUser(username, password, base_dn, fname, lname, domain, employee_num):
  """
  Create a new user account in Active Directory.
  """
  # LDAP connection
  try:
      ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, 0)
      ldap_connection = ldap.initialize(LDAP_SERVER)
      ldap_connection.simple_bind_s(BIND_DN, BIND_PASS)
  except ldap.LDAPError, error_message:
      print "Error connecting to LDAP server: %s" % error_message
      return False

  # Check and see if user exists
  try:
      user_results = ldap_connection.search_s(base_dn, ldap.SCOPE_SUBTREE,
                                              '(&(sAMAccountName=' +
                                              username +
                                              ')(objectClass=person))',
                                              ['distinguishedName'])
  except ldap.LDAPError, error_message:
      print "Error finding username: %s" % error_message
      return False

  # Check the results
  if len(user_results) != 0:
      print "User", username, "already exists in AD:", \
            user_results[0][1]['distinguishedName'][0]
      return False

  # Lets build our user: Disabled to start (514)
  user_dn = 'cn=' + fname + ' ' + lname + ',' + base_dn
  user_attrs = {}
  user_attrs['objectClass'] = \
            ['top', 'person', 'organizationalPerson', 'user']
  user_attrs['cn'] = fname + ' ' + lname
  user_attrs['userPrincipalName'] = username + '@' + domain
  user_attrs['sAMAccountName'] = username
  user_attrs['givenName'] = fname
  user_attrs['sn'] = lname
  user_attrs['displayName'] = fname + ' ' + lname
  user_attrs['userAccountControl'] = '514'
  user_attrs['mail'] = username + '@host.com'
  user_attrs['employeeID'] = employee_num
  user_attrs['homeDirectory'] = '\\\\server\\' + username
  user_attrs['homeDrive'] = 'H:'
  user_attrs['scriptPath'] = 'logon.vbs'
  user_ldif = modlist.addModlist(user_attrs)

  # Prep the password
  unicode_pass = unicode('\"' + password + '\"', 'iso-8859-1')
  password_value = unicode_pass.encode('utf-16-le')
  add_pass = [(ldap.MOD_REPLACE, 'unicodePwd', [password_value])]
  # 512 will set user account to enabled
  mod_acct = [(ldap.MOD_REPLACE, 'userAccountControl', '512')]
  # New group membership
  add_member = [(ldap.MOD_ADD, 'member', user_dn)]
  # Replace the primary group ID
  mod_pgid = [(ldap.MOD_REPLACE, 'primaryGroupID', GROUP_TOKEN)]
  # Delete the Domain Users group membership
  del_member = [(ldap.MOD_DELETE, 'member', user_dn)]

  # Add the new user account
  try:
      ldap_connection.add_s(user_dn, user_ldif)
  except ldap.LDAPError, error_message:
      print "Error adding new user: %s" % error_message
      return False

  # Add the password
  try:
      ldap_connection.modify_s(user_dn, add_pass)
  except ldap.LDAPError, error_message:
      print "Error setting password: %s" % error_message
      return False

  # Change the account back to enabled
  try:
      ldap_connection.modify_s(user_dn, mod_acct)
  except ldap.LDAPError, error_message:
      print "Error enabling user: %s" % error_message
      return False

  # Add user to their primary group
  try:
      ldap_connection.modify_s(GROUP_DN, add_member)
  except ldap.LDAPError, error_message:
      print "Error adding user to group: %s" % error_message
      return False

  # Modify user's primary group ID
  try:
      ldap_connection.modify_s(user_dn, mod_pgid)
  except ldap.LDAPError, error_message:
      print "Error changing user's primary group: %s" % error_message
      return False

  # Remove user from the Domain Users group
  try:
      ldap_connection.modify_s(DU_GROUP_DN, del_member)
  except ldap.LDAPError, error_message:
      print "Error removing user from group: %s" % error_message
      return False

  # LDAP unbind
  ldap_connection.unbind_s()

  # Setup user's home directory
  os.system('mkdir -p /home/' + username + '/public_html')
  os.system('cp /etc/skel/.bashrc /etc/skel/.bash_profile ' +
            '/etc/skel/.bash_logout /home/' + username)
  os.system('chown -R ' + username + ' /home/' + username)
  os.system('chmod 0701 /home/' + username)

  # All is good
  return True

More to come...