Consolidate external member code into two functions in baseldap.py

External members (users and hosts) are assumed when doing member
management on certain attributes. If the member isn't in IPA it
is assumed to be external. When doing member management we need
to sift through the list of failures and pull out all those
that were simply not found in IPA.

https://fedorahosted.org/freeipa/ticket/1734
This commit is contained in:
Rob Crittenden authored and Martin Kosek committed 2012-02-08 09:12:53 +01:00
1 parent cca80c7c8c
commit 304b70843a
5 files changed
+95 -283

No files matched your search

+9 -186
View File
@@ -74,7 +74,7 @@ class sudorule(LDAPObject):
object_name_plural = _('sudo rules')
object_class = ['ipaassociation', 'ipasudorule']
default_attributes = [
'cn', 'ipaenabledflag',
'cn', 'ipaenabledflag', 'externaluser',
'description', 'usercategory', 'hostcategory',
'cmdcategory', 'memberuser', 'memberhost',
'memberallowcmd', 'memberdenycmd', 'ipasudoopt',
@@ -393,30 +393,7 @@ class sudorule_add_user(LDAPAddMember):
return dn
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
completed_external = 0
# Sift through the user failures. We assume that these are all
# users that aren't stored in IPA, aka external users.
if 'memberuser' in failed and 'user' in failed['memberuser']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['externaluser'])
members = entry_attrs.get('memberuser', [])
external_users = entry_attrs_.get('externaluser', [])
failed_users = []
for user in failed['memberuser']['user']:
username = user[0].lower()
user_dn = self.api.Object['user'].get_dn(username)
if username not in external_users and user_dn not in members:
external_users.append(username)
completed_external += 1
else:
failed_users.append(username)
if completed_external:
try:
ldap.update_entry(dn, {'externaluser': external_users})
except errors.EmptyModlist:
pass
failed['memberuser']['user'] = failed_users
entry_attrs['externaluser'] = external_users
return (completed + completed_external, dn)
return add_external_post_callback('memberuser', 'user', 'externaluser', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_add_user)
@@ -428,28 +405,7 @@ class sudorule_remove_user(LDAPRemoveMember):
member_count_out = ('%i object removed.', '%i objects removed.')
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
# Run through the user failures and gracefully remove any defined as
# as an externaluser.
if 'memberuser' in failed and 'user' in failed['memberuser']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['externaluser'])
external_users = entry_attrs_.get('externaluser', [])
failed_users = []
completed_external = 0
for user in failed['memberuser']['user']:
username = user[0].lower()
if username in external_users:
external_users.remove(username)
completed_external += 1
else:
failed_users.append(username)
if completed_external:
try:
ldap.update_entry(dn, {'externaluser': external_users})
except errors.EmptyModlist:
pass
failed['memberuser']['user'] = failed_users
entry_attrs['externaluser'] = external_users
return (completed + completed_external, dn)
return remove_external_post_callback('memberuser', 'user', 'externaluser', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_remove_user)
@@ -470,30 +426,7 @@ class sudorule_add_host(LDAPAddMember):
return dn
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
completed_external = 0
# Sift through the host failures. We assume that these are all
# hosts that aren't stored in IPA, aka external hosts.
if 'memberhost' in failed and 'host' in failed['memberhost']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['externalhost'])
members = entry_attrs.get('memberhost', [])
external_hosts = entry_attrs_.get('externalhost', [])
failed_hosts = []
for host in failed['memberhost']['host']:
hostname = host[0].lower()
host_dn = self.api.Object['host'].get_dn(hostname)
if hostname not in external_hosts and host_dn not in members:
external_hosts.append(hostname)
completed_external += 1
else:
failed_hosts.append(hostname)
if completed_external:
try:
ldap.update_entry(dn, {'externalhost': external_hosts})
except errors.EmptyModlist:
pass
failed['memberhost']['host'] = failed_hosts
entry_attrs['externalhost'] = external_hosts
return (completed + completed_external, dn)
return add_external_post_callback('memberhost', 'host', 'externalhost', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_add_host)
@@ -505,29 +438,7 @@ class sudorule_remove_host(LDAPRemoveMember):
member_count_out = ('%i object removed.', '%i objects removed.')
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
# Run through the host failures and gracefully remove any defined as
# as an externalhost.
if 'memberhost' in failed and 'host' in failed['memberhost']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['externalhost'])
external_hosts = entry_attrs_.get('externalhost', [])
failed_hosts = []
completed_external = 0
for host in failed['memberhost']['host']:
hostname = host[0].lower()
if hostname in external_hosts:
external_hosts.remove(hostname)
completed_external += 1
else:
failed_hosts.append(hostname)
if completed_external:
try:
ldap.update_entry(dn, {'externalhost': external_hosts})
except errors.EmptyModlist:
pass
failed['memberhost']['host'] = failed_hosts
if external_hosts:
entry_attrs['externalhost'] = external_hosts
return (completed + completed_external, dn)
return remove_external_post_callback('memberhost', 'host', 'externalhost', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_remove_host)
@@ -568,30 +479,7 @@ class sudorule_add_runasuser(LDAPAddMember):
return dn
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
completed_external = 0
# Sift through the user failures. We assume that these are all
# users that aren't stored in IPA, aka external users.
if 'ipasudorunas' in failed and 'user' in failed['ipasudorunas']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['ipasudorunasextuser'])
members = entry_attrs.get('ipasudorunas', [])
external_users = entry_attrs_.get('ipasudorunasextuser', [])
failed_users = []
for user in failed['ipasudorunas']['user']:
username = user[0].lower()
user_dn = self.api.Object['user'].get_dn(username)
if username not in external_users and user_dn not in members:
external_users.append(username)
completed_external += 1
else:
failed_users.append(username)
if completed_external:
try:
ldap.update_entry(dn, {'ipasudorunasextuser': external_users})
except errors.EmptyModlist:
pass
failed['ipasudorunas']['user'] = failed_users
entry_attrs['ipasudorunasextuser'] = external_users
return (completed + completed_external, dn)
return add_external_post_callback('ipasudorunas', 'user', 'ipasudorunasextuser', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_add_runasuser)
@@ -603,28 +491,7 @@ class sudorule_remove_runasuser(LDAPRemoveMember):
member_count_out = ('%i object removed.', '%i objects removed.')
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
# Run through the user failures and gracefully remove any defined as
# as an externaluser.
if 'ipasudorunas' in failed and 'user' in failed['ipasudorunas']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['ipasudorunasextuser'])
external_users = entry_attrs_.get('ipasudorunasextuser', [])
failed_users = []
completed_external = 0
for user in failed['ipasudorunas']['user']:
username = user[0].lower()
if username in external_users:
external_users.remove(username)
completed_external += 1
else:
failed_users.append(username)
if completed_external:
try:
ldap.update_entry(dn, {'ipasudorunasextuser': external_users})
except errors.EmptyModlist:
pass
failed['ipasudorunas']['user'] = failed_users
entry_attrs['ipasudorunasextuser'] = external_users
return (completed + completed_external, dn)
return remove_external_post_callback('ipasudorunas', 'user', 'ipasudorunasextuser', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_remove_runasuser)
@@ -660,30 +527,7 @@ class sudorule_add_runasgroup(LDAPAddMember):
return dn
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
completed_external = 0
# Sift through the group failures. We assume that these are all
# groups that aren't stored in IPA, aka external groups.
if 'ipasudorunasgroup' in failed and 'group' in failed['ipasudorunasgroup']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['ipasudorunasextgroup'])
members = entry_attrs.get('ipasudorunasgroup', [])
external_groups = entry_attrs_.get('ipasudorunasextgroup', [])
failed_groups = []
for group in failed['ipasudorunasgroup']['group']:
groupname = group[0].lower()
group_dn = self.api.Object['group'].get_dn(groupname)
if groupname not in external_groups and group_dn not in members:
external_groups.append(groupname)
completed_external += 1
else:
failed_groups.append(groupname)
if completed_external:
try:
ldap.update_entry(dn, {'ipasudorunasextgroup': external_groups})
except errors.EmptyModlist:
pass
failed['ipasudorunasgroup']['group'] = failed_groups
entry_attrs['ipasudorunasextgroup'] = external_groups
return (completed + completed_external, dn)
return add_external_post_callback('ipasudorunasgroup', 'group', 'ipasudorunasextgroup', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_add_runasgroup)
@@ -695,28 +539,7 @@ class sudorule_remove_runasgroup(LDAPRemoveMember):
member_count_out = ('%i object removed.', '%i objects removed.')
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
# Run through the group failures and gracefully remove any defined as
# as an external group.
if 'ipasudorunasgroup' in failed and 'group' in failed['ipasudorunasgroup']:
(dn, entry_attrs_) = ldap.get_entry(dn, ['ipasudorunasextgroup'])
external_groups = entry_attrs_.get('ipasudorunasextgroup', [])
failed_groups = []
completed_external = 0
for group in failed['ipasudorunasgroup']['group']:
groupname = group[0].lower()
if groupname in external_groups:
external_groups.remove(groupname)
completed_external += 1
else:
failed_groups.append(groupname)
if completed_external:
try:
ldap.update_entry(dn, {'ipasudorunasextgroup': external_groups})
except errors.EmptyModlist:
pass
failed['ipasudorunasgroup']['group'] = failed_groups
entry_attrs['ipasudorunasextgroup'] = external_groups
return (completed + completed_external, dn)
return remove_external_post_callback('ipasudorunasgroup', 'group', 'ipasudorunasextgroup', ldap, completed, failed, dn, entry_attrs, keys, options)
api.register(sudorule_remove_runasgroup)