Make the user plugin use baseldap classes.
This commit is contained in:
Pavel Zuna authored and 
Rob Crittenden committed
1 parent
be82f941d0
commit
fbd1ff40f9
1 file changed
+76
-230
+76
-230
@@ -21,24 +21,51 @@
|
||||
Users (Identity)
|
||||
"""
|
||||
|
||||
from ipalib import api, crud, errors
|
||||
from ipalib import Command, Object
|
||||
from ipalib import api, errors
|
||||
from ipalib import Flag, Int, Password, Str
|
||||
from ipalib import uuid
|
||||
|
||||
# parent DN
|
||||
_container_dn = api.env.container_user
|
||||
|
||||
# attributes displayed by default
|
||||
_default_attributes = [
|
||||
'uid', 'givenname', 'sn', 'homedirectory', 'loginshell'
|
||||
]
|
||||
from ipalib.plugins.baseldap import *
|
||||
|
||||
|
||||
class user(Object):
|
||||
class user(LDAPObject):
|
||||
"""
|
||||
User object.
|
||||
"""
|
||||
container_dn = api.env.container_user
|
||||
object_name = 'user'
|
||||
object_name_plural = 'users'
|
||||
object_class = ['posixaccount']
|
||||
object_class_config = 'ipauserobjectclasses'
|
||||
default_attributes = [
|
||||
'uid', 'givenname', 'sn', 'homedirectory', 'loginshell', 'ou',
|
||||
'telephonenumber', 'title',
|
||||
]
|
||||
uuid_attribute = 'ipauniqueid'
|
||||
attribute_names = {
|
||||
'uid': 'user id',
|
||||
'cn': 'full name',
|
||||
'givenname': 'first name',
|
||||
'sn': 'last name',
|
||||
'homedirectory': 'home directory',
|
||||
'loginshell': 'login shell',
|
||||
'krbprincipalname': 'kerberos principal',
|
||||
'krblastpwdchange': 'last password change',
|
||||
'krbpasswordexpiration': 'password expiration',
|
||||
'uidnumber': 'uid number',
|
||||
'gidnumber': 'gid number',
|
||||
'memberof group': 'member of groups',
|
||||
'memberof netgroup': 'member of netgroups',
|
||||
'memberof rolegroup': 'member of rolegroups',
|
||||
'memberof taskgroup': 'member of taskgroups',
|
||||
'ipauniqueid': 'unique identifier'
|
||||
}
|
||||
attribute_order = [
|
||||
'uid', 'cn', 'givenname', 'sn', 'title', 'telephonenumber', 'ou',
|
||||
'homedirectory', 'loginshell', 'uidnumber', 'gidnumber', 'gecos',
|
||||
'krbprincipalname', 'krblastpwdchange', 'krbpasswordexpiration',
|
||||
]
|
||||
attribute_members = {
|
||||
'memberof': ['group', 'netgroup', 'rolegroup', 'taskgroup'],
|
||||
}
|
||||
|
||||
takes_params = (
|
||||
Str('givenname',
|
||||
@@ -59,6 +86,7 @@ class user(Object):
|
||||
Str('gecos?',
|
||||
doc='GECOS field',
|
||||
default_from=lambda uid: uid,
|
||||
autofill=True,
|
||||
),
|
||||
Str('homedirectory?',
|
||||
cli_name='homedir',
|
||||
@@ -74,6 +102,7 @@ class user(Object):
|
||||
cli_name='principal',
|
||||
doc='Kerberos principal name',
|
||||
default_from=lambda uid: '%s@%s' % (uid, api.env.realm),
|
||||
autofill=True,
|
||||
),
|
||||
Str('mail?',
|
||||
cli_name='email',
|
||||
@@ -96,313 +125,130 @@ class user(Object):
|
||||
api.register(user)
|
||||
|
||||
|
||||
class user_add(crud.Create):
|
||||
class user_add(LDAPCreate):
|
||||
"""
|
||||
Create new user.
|
||||
"""
|
||||
|
||||
def execute(self, *args, **options):
|
||||
ldap = self.api.Backend.ldap2
|
||||
uid = args[0]
|
||||
|
||||
# build entry attributes
|
||||
entry_attrs = self.args_options_2_entry(*args, **options)
|
||||
|
||||
# build entry DN
|
||||
dn = ldap.make_dn(entry_attrs, 'uid', _container_dn)
|
||||
|
||||
# get configuration entry attributes
|
||||
def pre_callback(self, ldap, dn, entry_attrs, *keys, **options):
|
||||
config = ldap.get_ipa_config()[1]
|
||||
|
||||
# fill in required attributes
|
||||
entry_attrs['objectclass'] = config.get('ipauserobjectclasses')
|
||||
|
||||
# fill default values
|
||||
# uidNumber gets filled automatically by the DS dna_plugin
|
||||
entry_attrs.setdefault('loginshell', config.get('ipadefaultloginshell'))
|
||||
entry_attrs.setdefault('gecos', uid)
|
||||
entry_attrs.setdefault(
|
||||
'krbprincipalname', '%s@%s' % (uid, self.api.env.realm)
|
||||
)
|
||||
# hack so we can request separate first and last name in CLI
|
||||
entry_attrs.setdefault(
|
||||
'cn', '%s %s' % (entry_attrs['givenname'], entry_attrs['sn'])
|
||||
)
|
||||
full_name = '%s %s' % (entry_attrs['givenname'], entry_attrs['sn'])
|
||||
entry_attrs.setdefault('cn', full_name)
|
||||
if 'homedirectory' not in entry_attrs:
|
||||
# get home's root directory from config
|
||||
homes_root = config.get('ipahomesrootdir', '/home')[0]
|
||||
# build user's home directory based on his uid
|
||||
home_dir = '%s/%s' % (homes_root, uid)
|
||||
home_dir = '%s/%s' % (homes_root, keys[-1])
|
||||
home_dir = home_dir.replace('//', '/').rstrip('/')
|
||||
entry_attrs['homedirectory'] = home_dir
|
||||
|
||||
# we're adding new users to a default group, get it's DN and gidNumber
|
||||
# we're adding new users to a default group, get its gidNumber
|
||||
# get default group name from config
|
||||
def_primary_group = config.get('ipadefaultprimarygroup')
|
||||
# build the group's DN
|
||||
group_parent_dn = self.api.env.container_group
|
||||
group_rdn = ldap.make_rdn_from_attr('cn', def_primary_group)
|
||||
group_dn = ldap.make_dn_from_rdn(group_rdn, group_parent_dn)
|
||||
# try to retrieve the group's gidNumber
|
||||
group_dn = self.api.Object['group'].get_dn(def_primary_group)
|
||||
try:
|
||||
(group_dn, group_attrs) = ldap.get_entry(group_dn, ['gidnumber'])
|
||||
except errors.NotFound:
|
||||
error_msg = 'Default group for new users not found.'
|
||||
raise errors.NotFound(reason=error_msg)
|
||||
# fill default group's gidNumber
|
||||
entry_attrs['gidnumber'] = group_attrs['gidnumber']
|
||||
|
||||
entry_attrs['ipauniqueid'] = str(uuid.uuid1())
|
||||
return dn
|
||||
|
||||
# create user entry
|
||||
ldap.add_entry(dn, entry_attrs)
|
||||
|
||||
# add user to default group
|
||||
def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
|
||||
config = ldap.get_ipa_config()[1]
|
||||
# add the user we just created into the default primary group
|
||||
def_primary_group = config.get('ipadefaultprimarygroup')
|
||||
group_dn = self.api.Object['group'].get_dn(def_primary_group)
|
||||
ldap.add_entry_to_group(dn, group_dn)
|
||||
|
||||
# get user entry with created attributes for output
|
||||
return ldap.get_entry(dn, entry_attrs.keys())
|
||||
|
||||
def output_for_cli(self, textui, result, *args, **options):
|
||||
(dn, entry_attrs) = result
|
||||
uid = args[0]
|
||||
|
||||
textui.print_name(self.name)
|
||||
textui.print_attribute('dn', dn)
|
||||
textui.print_entry(entry_attrs)
|
||||
textui.print_dashed('Created user "%s".' % uid)
|
||||
return dn
|
||||
|
||||
api.register(user_add)
|
||||
|
||||
|
||||
class user_del(crud.Delete):
|
||||
class user_del(LDAPDelete):
|
||||
"""
|
||||
Delete user.
|
||||
"""
|
||||
|
||||
def execute(self, uid):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
if uid == 'admin':
|
||||
# FIXME: add a specific exception for this?
|
||||
def pre_callback(self, ldap, dn, *keys, **options):
|
||||
if keys[-1] == 'admin':
|
||||
raise errors.ExecutionError('Cannot delete user "admin".')
|
||||
return dn
|
||||
|
||||
# build entry DN
|
||||
rdn = ldap.make_rdn_from_attr('uid', uid)
|
||||
dn = ldap.make_dn_from_rdn(rdn, _container_dn)
|
||||
|
||||
# delete user entry
|
||||
ldap.delete_entry(dn)
|
||||
|
||||
# blog about it
|
||||
self.log.info('IPA: %s "%s"' % (self.name, uid))
|
||||
|
||||
# return something positive
|
||||
def post_callback(self, ldap, dn, *keys, **options):
|
||||
self.log.info('IPA: %s "%s"' % (self.name, keys[-1]))
|
||||
return True
|
||||
|
||||
def output_for_cli(self, textui, result, uid):
|
||||
textui.print_name(self.name)
|
||||
textui.print_dashed('Deleted user "%s".' % uid)
|
||||
|
||||
api.register(user_del)
|
||||
|
||||
|
||||
class user_mod(crud.Update):
|
||||
class user_mod(LDAPUpdate):
|
||||
"""
|
||||
Modify user.
|
||||
"""
|
||||
|
||||
def execute(self, uid, **options):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
# build entry attributes, don't include uid!
|
||||
entry_attrs = self.args_options_2_entry(*tuple(), **options)
|
||||
|
||||
# build entry DN
|
||||
rdn = ldap.make_rdn_from_attr('uid', uid)
|
||||
dn = ldap.make_dn_from_rdn(rdn, _container_dn)
|
||||
|
||||
# update user entry
|
||||
ldap.update_entry(dn, entry_attrs)
|
||||
|
||||
# get user entry with modified + default attributes for output
|
||||
return ldap.get_entry(dn, (entry_attrs.keys() + _default_attributes))
|
||||
|
||||
def output_for_cli(self, textui, result, uid, **options):
|
||||
(dn, entry_attrs) = result
|
||||
|
||||
textui.print_name(self.name)
|
||||
textui.print_attribute('dn', dn)
|
||||
textui.print_entry(entry_attrs)
|
||||
textui.print_dashed('Modified user "%s".' % uid)
|
||||
|
||||
api.register(user_mod)
|
||||
|
||||
|
||||
class user_find(crud.Search):
|
||||
class user_find(LDAPSearch):
|
||||
"""
|
||||
Search for users.
|
||||
"""
|
||||
|
||||
takes_options = (
|
||||
Flag('all',
|
||||
doc='retrieve all attributes?',
|
||||
),
|
||||
)
|
||||
|
||||
def execute(self, term, **options):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
# get list of search fields from config
|
||||
config = ldap.get_ipa_config()[1]
|
||||
search_fields = config.get('ipausersearchfields')[0].split(',')
|
||||
|
||||
# look for term in all search fields
|
||||
search_kw = self.args_options_2_entry(**options)
|
||||
if term:
|
||||
for f in search_fields:
|
||||
search_kw[f] = '%s' % term
|
||||
# build search filter
|
||||
filter = ldap.make_filter(search_kw, exact=False)
|
||||
|
||||
# select attributes we want to retrieve
|
||||
if options['all']:
|
||||
attrs_list = ['*']
|
||||
else:
|
||||
attrs_list = _default_attributes
|
||||
|
||||
# get matching entries
|
||||
try:
|
||||
(entries, truncated) = ldap.find_entries(
|
||||
filter, attrs_list, _container_dn, ldap.SCOPE_ONELEVEL
|
||||
)
|
||||
except errors.NotFound:
|
||||
(entries, truncated) = (tuple(), False)
|
||||
|
||||
return (entries, truncated)
|
||||
|
||||
def output_for_cli(self, textui, result, term, **options):
|
||||
(entries, truncated) = result
|
||||
|
||||
textui.print_name(self.name)
|
||||
for (dn, entry_attrs) in entries:
|
||||
textui.print_attribute('dn', dn)
|
||||
textui.print_entry(entry_attrs)
|
||||
textui.print_plain('')
|
||||
textui.print_count(
|
||||
len(entries), '%i user matched.', '%i users matched.'
|
||||
)
|
||||
if truncated:
|
||||
textui.print_dashed('These results are truncated.', below=False)
|
||||
textui.print_dashed(
|
||||
'Please refine your search and try again.', above=False
|
||||
)
|
||||
|
||||
api.register(user_find)
|
||||
|
||||
|
||||
class user_show(crud.Retrieve):
|
||||
class user_show(LDAPRetrieve):
|
||||
"""
|
||||
Display user.
|
||||
"""
|
||||
|
||||
takes_options = (
|
||||
Flag('all',
|
||||
doc='retrieve all attributes?',
|
||||
),
|
||||
)
|
||||
|
||||
def execute(self, uid, **options):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
# build entry DN
|
||||
rdn = ldap.make_rdn_from_attr('uid', uid)
|
||||
dn = ldap.make_dn_from_rdn(rdn, _container_dn)
|
||||
|
||||
# select attributes we want to retrieve
|
||||
if options['all']:
|
||||
attrs_list = ['*']
|
||||
else:
|
||||
attrs_list = _default_attributes
|
||||
|
||||
return ldap.get_entry(dn, attrs_list)
|
||||
|
||||
def output_for_cli(self, textui, result, uid, **options):
|
||||
(dn, entry_attrs) = result
|
||||
|
||||
textui.print_name(self.name)
|
||||
textui.print_attribute('dn', dn)
|
||||
textui.print_entry(entry_attrs)
|
||||
|
||||
api.register(user_show)
|
||||
|
||||
|
||||
class user_lock(Command):
|
||||
class user_lock(LDAPQuery):
|
||||
"""
|
||||
Lock user account.
|
||||
"""
|
||||
def execute(self, *keys, **options):
|
||||
ldap = self.obj.backend
|
||||
|
||||
takes_args = (
|
||||
Str('uid',
|
||||
cli_name='user',
|
||||
doc='login name',
|
||||
),
|
||||
)
|
||||
dn = self.obj.get_dn(*keys, **options)
|
||||
|
||||
def execute(self, uid):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
# build entry DN
|
||||
rdn = ldap.make_rdn_from_attr('uid', uid)
|
||||
dn = ldap.make_dn_from_rdn(rdn, _container_dn)
|
||||
|
||||
# lock!
|
||||
try:
|
||||
ldap.deactivate_entry(dn)
|
||||
except errors.AlreadyInactive:
|
||||
pass
|
||||
|
||||
# return something positive
|
||||
return True
|
||||
|
||||
def output_for_cli(self, textui, result, uid):
|
||||
def output_for_cli(self, textui, result, *keys, **options):
|
||||
textui.print_name(self.name)
|
||||
textui.print_dashed('Locked user "%s".' % uid)
|
||||
textui.print_dashed('Locked user "%s".' % keys[-1])
|
||||
|
||||
api.register(user_lock)
|
||||
|
||||
|
||||
class user_unlock(Command):
|
||||
class user_unlock(LDAPQuery):
|
||||
"""
|
||||
Unlock user account.
|
||||
"""
|
||||
def execute(self, *keys, **options):
|
||||
ldap = self.obj.backend
|
||||
|
||||
takes_args = (
|
||||
Str('uid',
|
||||
cli_name='user',
|
||||
doc='login name',
|
||||
),
|
||||
)
|
||||
dn = self.obj.get_dn(*keys, **options)
|
||||
|
||||
def execute(self, uid):
|
||||
ldap = self.api.Backend.ldap2
|
||||
|
||||
# build entry DN
|
||||
rdn = ldap.make_rdn_from_attr('uid', uid)
|
||||
dn = ldap.make_dn_from_rdn(rdn, _container_dn)
|
||||
|
||||
# unlock!
|
||||
try:
|
||||
ldap.activate_entry(dn)
|
||||
except errors.AlreadyActive:
|
||||
pass
|
||||
|
||||
# return something positive
|
||||
return True
|
||||
|
||||
def output_for_cli(self, textui, result, uid):
|
||||
def output_for_cli(self, textui, result, *keys, **options):
|
||||
textui.print_name(self.name)
|
||||
textui.print_dashed('Unlocked user "%s".' % uid)
|
||||
textui.print_dashed('Unlocked user "%s".' % keys[-1])
|
||||
|
||||
api.register(user_unlock)
|
||||
|
||||
Reference in new issue
Block a user