Friday, February 11, 2011

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 build
python 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://"
BIND_DN = ""
BIND_PASS = "password"
USER_FILTER = "(&(objectClass=person)(primaryGroupID=7235))"
USER_BASE = "ou=Special Peeps,ou=My Users,dc=host,dc=com"

# LDAP connection
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')

# Lookup usernames from LDAP via paged search
paged_results_control = SimplePagedResultsControl(
accounts = []
pages = 0
while True:
serverctrls = [paged_results_control]
msgid = ldap_connection.search_ext(USER_BASE,
except ldap.LDAPError, e:
sys.stderr.write('Error performing user paged search: ' +
str(e) + '\n')
unused_code, results, unused_msgid, serverctrls = \
except ldap.LDAPError, e:
sys.stderr.write('Error getting user paged search results: ' +
str(e) + '\n')
for result in results:
pages += 1
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)
if not cookie:

# LDAP unbind

# Make dictionary with user data
user_map = {}
for entry in accounts:
if entry[1].has_key('employeeID') and \
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
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')

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
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
user_results = ldap_connection.search_s(base_dn, ldap.SCOPE_SUBTREE,
'(&(sAMAccountName=' +
username +
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:", \
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 + ''
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
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
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
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
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
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
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

# 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...