You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
tde-guidance/userconfig/unixauthdb.py

1155 lines
39 KiB

#!/usr/bin/python3
###########################################################################
# Copyright (C) 2004-2006 by Simon Edwards
# <simon@simonzone.com>
#
# Copyright: See COPYING file that comes with this distribution
#
###########################################################################
# An API for querying and modifying the authorisation database on Unix systems.
#
# The first function that you need to use is getContext(). It returns a
# Context object that contains all relevant information concerning
# the current authorisation database on this machine.
import crypt
import random
import fcntl
import time
import os
import os.path
import stat
import shutil
import codecs
import locale
import tempfile
ldaperror = ""
try:
import ldap
except ImportError:
ldaperror = "The LDAP Python Module is not installed, but needed to use LDAP. Install it."
def createTempFile(origfile):
origstat = os.stat(origfile)
tmp_prefix = os.path.basename(origfile) + "."
tmp_dir = os.path.dirname(origfile)
try:
ret = tempfile.mkstemp(prefix=tmp_prefix, dir=tmp_dir)
except:
raise IOError("Unable to create a new temporary file for " + origfile)
(fd, tmpfile) = ret
shutil.copymode(origfile, tmpfile)
os.chown(tmpfile, origstat.st_uid, origstat.st_gid)
return ret
def getContext(editmode=False):
"""Get a Context object describing the system's authorisation database.
Parameters:
editmode - Set to true if you also wish change the information in this
context. Root access is required. Defaults to false.
Returns a Context object.
If the environmental variable "USERCONFIG_USES_LDAP" is set to "true",
userconfig will use LDAP as the backend. This feature is in development
and using it is not recommended, it won't work.
"""
# Detect what kind of auth system we are running on and create
# and initialise the corresponding Context object type.
# Check for Mandrake
# Check libuser.conf
try:
if os.environ["USERCONFIG_USES_LDAP"].lower() == "true":
use_ldap = True
except KeyError as e:
use_ldap = False
if not use_ldap:
return PwdContext(editmode)
else:
print("===================================================================")
print("Warning:")
print("\tYou are using LDAP as backend. This feature is under development")
print("\tand it is currently not recommended to use it.")
print("\tIf you do not want to use LDAP as backend, set the environmental")
print("\tvariabale 'USERCONFIG_USES_LDAP' to 'False'.")
print("===================================================================")
return LdapContext(editmode)
###########################################################################
# Base classes.
#
class Context(object):
"""Contains all of the information about the current authorisation
database, plus some useful methods for modify this information.
"""
def __init__(self):
self._users = []
self._groups = []
self._shells = None
self._setDefaultValues()
def newUser(self,defaults=False,systemuser=False):
"""Create a new UnixUser object.
Creates a new blank UnixUser object. The object is not part of the
current Context. You need to add it yourself using addUser().
Newly allocated UIDs are unique with respect to the list of UnixUser
objects in the Context.
Keyword arguments:
defaults -- Set to true if the new object should be filled in with
reasonable default values for the UID and username.
(default False)
systemuser -- Should the new user be allocated a UID from the system
range of UIDs. (default is False)
Returns a new UnixUser object.
"""
newuserobj = self._createUser()
if defaults:
if systemuser:
r = range(0,self.last_system_uid)
else:
r = range(self.first_uid,self.last_uid)
for candiate in r:
for u in self._users:
if u.getUID()==candiate:
break
else:
newuserobj.setUID(candiate)
break
if self.lookupUsername('new_user') is None:
newuserobj.setUsername('new_user')
else:
i = 1
while 1:
if self.lookupUsername('new_user_'+str(i)) is None:
newuserobj.setUsername('new_user_'+str(i))
break
i += 1
return newuserobj
def getUsers(self):
"""Get a list of all existing users.
Returns an array of UnixUser objects.
"""
#print "USERS:", self._users
return self._users[:]
def getGroups(self):
"""Get a list of all existing groups.
Returns an array of UnixGroup objects.
"""
try:
self._groups.remove("new_user")
except ValueError:
print("no user removed")
pass
return self._groups[:]
def newGroup(self,defaults=False,systemgroup=False):
"""Create a new UnixGroup object.
Creates a new blank UnixGroup object. The object is not part of the
current Context. You need to add it yourself using addGroup().
Newly allocated GIDs are unique with respect to the list of UnixGroup
objects in the Context.
Keyword arguments:
defaults -- Set to true if the new object should be filled in with
reasonable default values for the GID and groupname.
(default False)
systemgroup -- Set to True if the newly allocated GID should come
from the pool of system group IDs. (default False)
Returns a new UnixGroup object.
"""
newgroupobj = self._createGroup()
if defaults:
if systemgroup:
r = range(0,self.last_system_gid)
else:
r = range(self.first_gid,self.last_gid)
for candiate in r:
for u in self._groups:
if u.getGID()==candiate:
break
else:
newgroupobj.setGID(candiate)
break
if self.lookupGroupname('new_group') is None:
newgroupobj.setGroupname('new_group')
else:
i = 1
while 1:
if self.lookupGroupname('new_user_'+str(i)) is None:
newgroupobj.setGroupname('new_user_'+str(i))
break
i += 1
return newgroupobj
def _createGroup(self):
raise NotImplementedError("Context.newGroup()")
def addUser(self,userobj):
"""Adds the given user to the authorisation database.
This change only takes effect after calling context.save().
Keyword arguments:
userobj -- The UnixUser object to add.
"""
self._users.append(userobj)
def addGroup(self,groupobj):
"""Adds the given group to the authorisation database.
This change only takes effect after calling context.save().
Keyword arguments:
groupobj -- The UnixGroup object to add.
"""
if groupobj not in self._groups:
self._groups.append(groupobj)
def removeUser(self,userobj):
"""Removes the given user object from the authorisation database.
The user is also removed from all groups.
This change only takes effect after calling context.save().
"""
for g in userobj.getGroups():
userobj.removeFromGroup(g)
self._users.remove(userobj)
def removeGroup(self,groupobj):
"""Removes the given group object from the authorisation database.
All users are removed from the group.
This change only takes effect after calling context.save().
"""
for u in groupobj.getUsers():
u.removeFromGroup(groupobj)
self._groups.remove(groupobj)
def lookupUID(self,uid):
"""Lookup a UnixUser object by its numeric user ID.
Keyword arguments:
uid -- User ID to lookup, integer.
Returns the matching UnixUser object or None if it was not found.
"""
for user in self._users:
if user.getUID()==uid:
return user
return None
def lookupUsername(self,username):
"""Lookup a UnixUser object by username.
Keyword arguments:
username -- Username to lookup, string.
Returns the matching UnixUser object or None if it was not found.
"""
for user in self._users:
if user.getUsername()==username:
return user
return None
def lookupGID(self,gid):
"""Lookup a UnixGroup object by its numeric group ID.
Keyword arguments:
gid -- Group ID to lookup, integer.
Returns the matching UnixGroup object or None if it was not found.
"""
for group in self._groups:
if group.getGID()==gid:
return group
return None
def lookupGroupname(self,groupname):
"""Lookup a UnixGroup object by groupname.
Returns the matching UnixGroup object or None if it was not found.
"""
for group in self._groups:
if group.getGroupname()==groupname:
return group
return None
def getUserShells(self):
"""Get the list of available login shells.
Returns an array of strings.
"""
if self._shells is None:
self._shells = []
fhandle = codecs.open('/etc/shells','r',locale.getpreferredencoding())
for l in fhandle.readlines():
# TODO: strangely this lets some comented lines slip through
if len(l.strip()) > 1 and l.strip()[0] != "#":
# Only show existing shells
if os.path.isfile(l.strip()):
self._shells.append(l.strip())
fhandle.close()
return self._shells[:]
def save(self):
"""Synchronises the Context with the underlying operating system.
After a successful save, any changes to the Context will be reflected
system wide.
"""
raise NotImplementedError("Context.save()")
def createHomeDirectory(self,userobj):
if os.path.exists(userobj.getHomeDirectory()):
raise IOError("Home directory %s already exists." % userobj.getHomeDirectory())
# Copy the skeleton directory over
shutil.copytree(self._getSkeletonDirectory(),userobj.getHomeDirectory(),True)
# Fix the file ownership stuff
uid = userobj.getUID()
gid = userobj.getPrimaryGroup().getGID()
os.chmod(userobj.getHomeDirectory(),self.dir_mode)
#os.system("chmod "+self.dir_mode+" "+userobj.getHomeDirectory())
#print "Setting permissions:", userobj.getHomeDirectory(),self.dir_mode
os.lchown(userobj.getHomeDirectory(),uid,gid)
for root,dirs,files in os.walk(userobj.getHomeDirectory()):
for d in dirs:
os.lchown(os.path.join(root,d),uid,gid)
for f in files:
os.lchown(os.path.join(root,f),uid,gid)
def removeHomeDirectory(self,userobj):
if os.path.exists(userobj.getHomeDirectory()):
shutil.rmtree(userobj.getHomeDirectory())
def _createUser(self):
raise NotImplementedError("Context._createUser()")
def _sanityCheck(self):
userids = []
for u in self._users:
if isinstance(u,UnixUser)==False:
raise TypeError("Found an object in the list of users that is not a UnixUser object.")
uid = u.getUID()
if uid in userids:
raise ValueError("User ID %i appears more than once." % uid)
userids.append(uid)
u._sanityCheck()
groupids = []
for g in self._groups:
if isinstance(g,UnixGroup)==False:
raise TypeError("Found an object in the list of groups that is not a UnixGroup object.")
gid = g.getGID()
if gid in groupids:
raise ValueError("Group ID %i appears more than once." % gid)
groupids.append(gid)
g._sanityCheck()
def _getSkeletonDirectory(self):
return self.skel
def _readAdduserConf(self):
""" Fill a dictionary with the values from /etc/adduser.conf
which then can be used as default values, if the file exists
at least.
Attention: We're not validating!"""
self.defaults = {}
self.adduserconf = '/etc/adduser.conf'
if not os.path.isfile(self.adduserconf):
return
fhandle = codecs.open(self.adduserconf,'r',locale.getpreferredencoding())
for line in fhandle.readlines():
line = line.strip()
parts = line.split("=")
if len(parts) == 2:
self.defaults[str(parts[0].strip())] = parts[1].strip()
def _setDefaultValues(self):
""" Set a lot of default values for UIDs and GIDs, try to use the values
from /etc/adduser.conf."""
self._readAdduserConf()
try:
self.skel = self.defaults["SKEL"]
except KeyError:
self.skel = '/etc/skel'
# IDs for new users and groups.
try:
self.first_uid = int(self.defaults['FIRST_UID'])
except (KeyError,ValueError):
self.first_uid = 1000
try:
self.last_uid = int(self.defaults["LAST_UID"])
except (KeyError,ValueError):
self.last_uid = 29999
try:
self.first_gid = int(self.defaults["FIRST_GID"])
except (KeyError,ValueError):
self.first_gid = 1000
try:
self.last_gid = int(self.defaults["LAST_GID"])
except (KeyError,ValueError):
self.last_gid = 65534
# Which IDs are system user and system groups?
try:
self.first_system_uid = int(self.defaults["FIRST_SYSTEM_UID"])
except (KeyError,ValueError):
self.first_system_uid = 500
try:
self.last_system_uid = int(self.defaults["LAST_SYSTEM_UID"])
except (KeyError,ValueError):
self.last_system_uid = 65534
try:
self.first_system_gid = int(self.defaults["FIRST_SYSTEM_GID"])
except (KeyError,ValueError):
self.first_system_gid = 500
try:
self.last_system_gid = int(self.defaults["LAST_SYSTEM_GID"])
except (KeyError,ValueError):
self.last_system_gid = 65534
# More defaults which might make sense.
try:
self.dir_mode = int(self.defaults["DIR_MODE"],8)
except (KeyError,ValueError):
self.dir_mode = int("0755",8)
print("Didn't read default DIR_MODE")
try:
self.dhome = self.defaults["DHOME"]
except KeyError:
self.dhome = "/home"
try:
self.dshell = self.defaults["DSHELL"]
except KeyError:
# Will be set in showNewUser()
self.dshell = None
###########################################################################
class UnixUser(object):
def __init__(self,context):
self._context = context
self._uid = None
self._username = None
# UnixGroup object.
self._primarygroup = None
# List of UnixGroup objects.
self._groups = []
self._gecos = None
self._homedirectory = None
self._loginshell = None
self._islocked = False
self._encpass = ""
# FIXME : This should actually be days since epoch or something like this
self._passlastchange = 0
self._passminimumagebeforechange = 0
self._passmaximumage = None
self._passexpirewarn = 7
self._passexpiredisabledays = None
self._disableddays = None
def polish(self):
primary_group = self._context.lookupGID(self._gid)
if primary_group is None:
# The GID didn't match an existing group. Quickly make a new group.
new_group = self._context.newGroup()
new_group.setGID(self._gid)
new_group_name = "group%i" % self._gid
i = 0
while self._context.lookupGroupname(new_group_name) is not None:
i += 1
new_group_name = "group%i_%i" % (self._gid,i)
new_group.setGroupname(new_group_name)
self._context.addGroup(new_group)
primary_group = new_group
self.setPrimaryGroup(primary_group)
for group in self._context._groups:
if group.contains(self):
self._groups.append(group)
def getUID(self):
"""Get the unix user ID.
Returns the integer.
"""
return self._uid
def setUID(self,uid):
"""Set the unix user ID.
Keyword arguments:
uid -- Integer user id.
"""
uid = int(uid)
if uid<0:
raise ValueError("User ID (%i) is a negative number." % uid)
self._uid = uid
def isSystemUser(self):
"""See if this user is a system user.
Returns True or False.
"""
return not (self._context.first_uid <= self._uid < self._context.last_uid)
def getUsername(self): return self._username
def setUsername(self,username): self._username = username
def getPrimaryGroup(self):
"""Get the primary group for this user.
Returns a UnixGroup object.
"""
return self._primarygroup
def setPrimaryGroup(self,groupobj):
"""Set the primary group for this user.
If the given group is not part of this user's list of groups, then
it will be added.
Keyword arguments:
groupobj -- The group to set as the primary group.
"""
self.addToGroup(groupobj)
self._primarygroup = groupobj
def getGroups(self):
"""Get the list of groups that this user belongs to.
The user's primary group is also included in the returned list.
Returns a list of UnixGroup objects. Modify the list does not affect
this UnixUser object.
"""
return self._groups[:]
def addToGroup(self,groupobj):
"""Add this user to the given group.
Keyword arguments:
groupobj -- UnixGroup object.
"""
groupobj._addUser(self)
if groupobj not in self._groups:
self._groups.append(groupobj)
def removeFromGroup(self,groupobj):
"""Remove this user from the given group.
If group is current this user's primary group, then
Keyword arguments:
groupobj -- UnixGroup object.
"""
groupobj._removeUser(self)
try:
self._groups.remove(groupobj)
except ValueError:
pass
if self._primarygroup is groupobj:
if len(self._groups)==0:
self._primarygroup = None
else:
self._primarygroup = self._groups[0]
def setPassword(self,password):
# Make some salt.
space = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQSRTUVWXYZ0123456789./'
salt = ""
for x in range(8):
salt += space[random.randint(0,len(space)-1)]
self._encpass = crypt.crypt(password,'$1$'+salt+'$')
def isLocked(self): return self._islocked
def setLocked(self,locked): self._islocked = locked
def getRealName(self):
if not self._gecos:
return ""
try:
return self._gecos.split(",")[0]
except AttributeError:
return self._gecos
def setRealName(self,realname): self._gecos = realname
def getHomeDirectory(self): return self._homedirectory
def setHomeDirectory(self,homedirectory): self._homedirectory = homedirectory
def getLoginShell(self): return self._loginshell
def setLoginShell(self,loginshell): self._loginshell = loginshell
# 'None' means that there is no maximum password age.
def getMaximumPasswordAge(self): return self._passmaximumage
def setMaximumPasswordAge(self,days): self._passmaximumage = days
def getMinimumPasswordAgeBeforeChange(self): return self._passminimumagebeforechange
def setMinimumPasswordAgeBeforeChange(self,days): self._passminimumagebeforechange = days
def getPasswordDisableAfterExpire(self): return self._passexpiredisabledays
def setPasswordDisableAfterExpire(self,days): self._passexpiredisabledays = days
def getPasswordExpireWarning(self): return self._passexpirewarn
def setPasswordExpireWarning(self,days): self._passexpirewarn = days
def getLastPasswordChange(self): return self._passlastchange
def getExpirationDate(self): return self._disableddays
def setExpirationDate(self,unixdate): self._disableddays = unixdate
def __str__(self):
return "%s(%i)" % (self._username,self._uid)
def _sanityCheck(self):
if self._primarygroup is None:
raise ValueError("Userobj has no primary group!")
if self._uid is None:
raise ValueError("Userobj has no UID!")
###########################################################################
class UnixGroup(object):
def __init__(self,context):
self._context = context
# List of UnixUser objects.
self._members = []
self._gid = None
self._groupname = None
def contains(self,userobj):
"""Check if a the given user is a member of this group.
Returns True or False.
"""
return userobj in self._members
def polish(self): pass
def isSystemGroup(self):
"""Check if this group is a system group.
Returns True or False.
"""
return not (self._context.first_gid <= self._gid < self._context.last_gid)
#return not (500 <= self._gid < 65534)
def getGID(self):
"""Get the unix group ID.
Returns the integer group id.
"""
return self._gid
def setGID(self,gid):
"""Set the unix group ID.
Keyword arguments:
gid -- new group id, integer.
"""
self._gid = gid
def getGroupname(self): return self._groupname
def setGroupname(self,groupname): self._groupname = groupname
def getUsers(self): return self._members[:]
def _addUser(self,userobj):
if not self.contains(userobj):
self._members.append(userobj)
def _removeUser(self,userobj):
try:
self._members.remove(userobj)
except ValueError:
pass
def __str__(self):
# FIXME encoding
return str(self._groupname) + " (" + str(self._gid) + ") " + str([str(u) for u in self._members])
def _sanityCheck(self):
pass
###########################################################################
class PwdContext(Context):
#def __init__(self,editmode,passwordfile="etc-passwd",groupfile='etc-group',shadowfile="etc-shadow"):
def __init__(self,editmode,passwordfile="/etc/passwd",groupfile='/etc/group',shadowfile="/etc/shadow"):
Context.__init__(self)
self.__editmode = editmode
self.__passwordfile = passwordfile
self.__groupfile = groupfile
self.__shadowfile = shadowfile
self._setDefaultValues()
# Read in the password file
fhandle = codecs.open(passwordfile,'r',locale.getpreferredencoding())
if LockFDRead(fhandle.fileno())==False:
raise IOError("Unable to lock the "+passwordfile+" file.")
try:
for line in fhandle.readlines():
if line.strip()!="":
newuserobj = self.newUser(False)
newuserobj._initString(line)
self._users.append(newuserobj)
finally:
UnlockFD(fhandle.fileno())
fhandle.close()
# Read the group file
fhandle = codecs.open(groupfile,'r',locale.getpreferredencoding())
if LockFDRead(fhandle.fileno())==False:
raise IOError("Unable to lock the "+groupfile+" file.")
try:
for line in fhandle.readlines():
if line.strip()!="":
newgroupobj = self.newGroup(False)
newgroupobj._initString(line)
self._groups.append(newgroupobj)
finally:
UnlockFD(fhandle.fileno())
fhandle.close()
if self.__editmode:
# Load up the info from the shadow file too.
fhandle = codecs.open(shadowfile,'r',locale.getpreferredencoding())
if LockFDRead(fhandle.fileno())==False:
raise IOError("Unable to lock the "+shadowfile+" file.")
try:
for line in fhandle.readlines():
if line.strip()!="":
try:
(username,encpass,passlastchange,passminimumagebeforechange,passmaximumage, \
passexpirewarn,passexpiredisabledays,disableddays,reserve) = \
tuple(line.strip().split(":"))
userobj = self.lookupUsername(username)
if userobj is not None:
if encpass=="":
encpass = "*"
userobj._encpass = encpass
if userobj._encpass[0]=='!':
userobj._islocked = True
userobj._encpass = userobj._encpass[1:]
else:
userobj._islocked = False
# FIXME : set time
if passlastchange and passlastchange!="None":
userobj._passlastchange = int(passlastchange)
else:
passlastchange = 0
if passminimumagebeforechange=="":
passminimumagebeforechange = None
else:
passminimumagebeforechange = int(passminimumagebeforechange)
if passminimumagebeforechange>=99999:
passminimumagebeforechange = None
userobj._passminimumagebeforechange = passminimumagebeforechange
if passmaximumage=="":
passmaximumage = None
else:
passmaximumage = int(passmaximumage)
if passmaximumage>=99999:
passmaximumage = None
userobj._passmaximumage = passmaximumage
if passexpirewarn=="":
passexpirewarn = None
else:
passexpirewarn = int(passexpirewarn)
if passexpirewarn>=99999:
passexpirewarn = None
userobj._passexpirewarn = passexpirewarn
if passexpiredisabledays=="":
userobj._passexpiredisabledays = None
else:
userobj._passexpiredisabledays = int(passexpiredisabledays)
if disableddays=="" or disableddays=="99999":
userobj._disableddays = None
else:
userobj._disableddays = int(disableddays)
userobj._reserve = reserve
else:
print("Couldn't find",username)
except ValueError:
pass
finally:
UnlockFD(fhandle.fileno())
fhandle.close()
for group in self._groups:
group.polish()
for user in self._users:
user.polish()
def _createUser(self):
return PwdUser(self)
def _createGroup(self):
return PwdGroup(self)
def save(self):
if self.__editmode==False:
raise IOError("Can't save, the context was created Read only.")
self._sanityCheck()
# Write out the new password file.
(fd, tmpname) = createTempFile(self.__passwordfile)
for u in self._users:
os.write(fd, u._getPasswdEntry().encode(locale.getpreferredencoding(),'replace'))
#print u._getPasswdEntry()
os.close(fd)
# Update the passwd file
passwordlock = os.open(self.__passwordfile, os.O_WRONLY) # FIXME encoding
if LockFDWrite(passwordlock)==False:
raise IOError("Couldn't get a write lock on "+self.__passwordfile)
try:
os.rename(tmpname, self.__passwordfile)
finally:
UnlockFD(passwordlock)
os.close(passwordlock)
# Write out the new group file
(fd, tmpname) = createTempFile(self.__groupfile)
origstat = os.stat(self.__groupfile)
for g in self._groups:
os.write(fd,g._getGroupFileEntry().encode(locale.getpreferredencoding()))
#print g._getGroupFileEntry()[:-1]
os.close(fd)
os.chown(tmpname, origstat.st_uid, origstat.st_gid)
# Update the group file.
grouplock = os.open(self.__groupfile, os.O_WRONLY)
if LockFDWrite(grouplock)==False:
raise IOError("Couldn't get write lock on "+self.__groupfile)
try:
os.rename(tmpname, self.__groupfile)
finally:
UnlockFD(grouplock)
os.close(grouplock)
# Write out the new shadow file
origstat = os.stat(self.__shadowfile)
(fd, tmpname) = createTempFile(self.__shadowfile)
for u in self._users:
os.write(fd,u._getShadowEntry().encode(locale.getpreferredencoding()))
#print u._getShadowEntry()[:-1]
os.close(fd)
# Update the shadow file.
# Make sure that it is writable.
if (origstat.st_mode & stat.S_IWUSR)==0:
os.chmod(self.__shadowfile,origstat.st_mode|stat.S_IWUSR)
shadowlock = os.open(self.__shadowfile, os.O_WRONLY)
if LockFDWrite(shadowlock)==False:
raise IOError("Couldn't get write lock on "+self.__shadowfile)
try:
os.rename(tmpname, self.__shadowfile)
finally:
UnlockFD(shadowlock)
os.close(shadowlock)
# set the permissions back to thier default.
if (origstat.st_mode & stat.S_IWUSR)==0:
os.chmod(self.__shadowfile,origstat.st_mode)
###########################################################################
class LdapContext(Context):
def __init__(self,editmode,server="localhost",admin_dn="",admin_pass=""):
""" Connect to the LDAP server and invoke further actions.
"""
Context.__init__(self)
# admin_dn is DistinguishedName? (or dn, for short)
self.server = server
self.baseDN = "dc=vizZzion,dc=net"
self.url = "ldap://"+self.server
self.ldapserver = ldap.initialize(self.url)
self.ldapserver.protocol_version = ldap.VERSION3
self.editmode = editmode
if not self.editmode:
self.ldapserver.simple_bind("admin",admin_pass)
print("Connected to ", self.url)
self._users = self._getUsers()
def _getUsers(self):
""" Retrieve a list of users from the LDAP server.
"""
_users = []
print("LdapContext._getUsers")
searchScope = ldap.SCOPE_SUBTREE
retrieveAttributes = None
searchFilter = "cn=*"
try:
ldap_result_id = self.ldapserver.search(self.baseDN, searchScope, searchFilter, retrieveAttributes)
result_set = []
while 1:
result_type, result_data = self.ldapserver.result(ldap_result_id, 0)
if (result_data == []):
break
else:
if result_type == ldap.RES_SEARCH_ENTRY:
#print result_data[0][1]
#print " --------------------- "
result_set.append(result_data[0][1])
#print result_set
except ldap.LDAPError as e:
print("ERROR: ",e)
if len(result_set) == 0:
print("No Results.")
return
count = 0
"""
for entry in result_set:
for d in entry.keys():
print d, "::", entry[d]
print "======== Next User =============="
"""
# Walk through result_set and create users.
for entry in result_set:
try:
name = entry['cn'][0]
login = entry['uid'][0]
loginshell = entry['loginShell'][0]
homedirectory = entry['homeDirectory'][0]
uid = entry['uidNumber'][0]
gid = entry['gidNumber'][0]
count = count + 1
#print "\n%d. User: %s\n\tName: %s\n\tShell: %s\n\tHomeDir: %s\n\tUID: %s\n\tGID: %s\n" %\
# (count, login, name, loginshell, homedirectory, uid, gid)
# Create a new userobject
new_user = self._createUser()
new_user.setHomeDirectory(homedirectory)
new_user.setUID(uid)
new_user.setRealName(name)
new_user.setLoginShell(loginshell)
new_user.setUsername(login)
_users.append(new_user)
print("Number of Users:", len(self._users))
except KeyError as e:
# Debugging output...
print("ERR:: ",e)
print('err:: ',entry)
return _users
def _createUser(self):
return LdapUser(self)
def _createGroup(self):
return LdapGroup(self)
def save(self):
print("LdapContext.save() does nothing yet.")
###########################################################################
class LdapUser(UnixUser):
def __str__(self):
return "LdapUser: %s(%i)" % (self._username,self._uid)
###########################################################################
class LdapGroup(UnixGroup):
def __str__(self):
return "LdapGroup: %s(%i)" % (self._username,self._uid)
###########################################################################
class PwdUser(UnixUser):
def __init__(self,context):
UnixUser.__init__(self,context)
self._reserve = ""
def _initString(self,line):
(self._username,x,self._uid,self._gid,self._gecos,self._homedirectory, \
self._loginshell) = tuple(line.strip().split(":"))
self._uid = int(self._uid)
self._gid = int(self._gid)
def _getPasswdEntry(self):
return ":".join( [self._username,
"x",
str(self._uid),
str(self._primarygroup.getGID()),
self._gecos,
self._homedirectory,
self._loginshell ] ) + "\n"
def _getShadowEntry(self):
if self._islocked:
encpass = '!' + self._encpass
else:
encpass = self._encpass
if self._passminimumagebeforechange==None:
passminimumagebeforechange = ""
else:
passminimumagebeforechange = str(self._passminimumagebeforechange)
if self._passmaximumage==None:
passmaximumage = "99999"
else:
passmaximumage = str(self._passmaximumage)
if self._disableddays==None:
disableddays = ""
else:
disableddays = str(self._disableddays)
if self._passexpiredisabledays==None:
passexpiredisabledays = ""
else:
passexpiredisabledays = str(self._passexpiredisabledays)
if self._passexpirewarn==None:
passexpirewarn = ""
else:
passexpirewarn = str(self._passexpirewarn)
return ":".join( [self._username,
encpass,
str(self._passlastchange),
passminimumagebeforechange,
passmaximumage,
passexpirewarn,
passexpiredisabledays,
disableddays,
self._reserve ])+ "\n"
###########################################################################
class PwdGroup(UnixGroup):
def __init__(self,context):
UnixGroup.__init__(self,context)
self._memberids = ""
self._encpass = ""
def _initString(self,line):
(self._groupname,self._encpass,self._gid,self._memberids) = tuple(line.strip().split(":"))
self._gid = int(self._gid)
def polish(self):
membernames = self._memberids.split(",")
for username in membernames:
userobj = self._context.lookupUsername(username)
if userobj!=None:
self._members.append(userobj)
def _getGroupFileEntry(self):
return ":".join( [ self._groupname,
self._encpass,
str(self._gid),
",".join([u.getUsername() for u in self._members if u.getPrimaryGroup() is not self])]) + "\n"
###########################################################################
def LockFDRead(fd):
retries = 6
while retries!=0:
try:
fcntl.lockf(fd,fcntl.LOCK_SH | fcntl.LOCK_NB)
return True
except IOError:
# Wait a moment
time.sleep(1)
return False
def LockFDWrite(fd):
retries = 6
while retries!=0:
try:
fcntl.lockf(fd,fcntl.LOCK_EX | fcntl.LOCK_NB)
return True
except IOError:
# Wait a moment
time.sleep(1)
return False
def UnlockFD(fd):
fcntl.lockf(fd,fcntl.LOCK_UN)
###########################################################################
if __name__=='__main__':
print("Testing")
context = getContext(True)
print("Stopping here...")
#import sys
#sys.exit(0) ## Remove.
#print "Users:"
#for user in context.getUsers():
for user in context._users:
print("--------------------------------------------------")
print("UID:",user.getUID())
print("Is system user:",user.isSystemUser())
print("Username:",user.getUsername())
print("Primary Group:",str(user.getPrimaryGroup()))
print("Groups:",[str(u) for u in user.getGroups()])
print("Is locked:",user.isLocked())
print("Real name:",user.getRealName())
print("Home Dir:",user.getHomeDirectory())
print("Maximum password age:",user.getMaximumPasswordAge())
print("Minimum password age before change:",user.getMinimumPasswordAgeBeforeChange())
print("Expire warning:",user.getPasswordExpireWarning())
print("Disable after Expire:",user.getPasswordDisableAfterExpire())
#print user._getPasswdEntry()
print("Groups")
for group in context.getGroups():
print(str(group))
#print group._getGroupFileEntry()
print("Saving")
context.save()