IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Fix netgroup plugin to use correct member attribute names.

Browse Source 
When the netgroup plugin was rebased it ended up using the member
attribute for its memberships and not memberuser/memberhost.

I also fixed this same attribute problem in the tests and tried to beef
them up a little. If nis/schema compat are enabled it will try to compare
the generated triplets with a known-good value.
This commit is contained in:
Rob Crittenden
2010-07-14 14:45:15 -04:00
parent 0d12b0344f
commit 57a9001f8d
2 changed files with 148 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
ipalib/plugins/netgroup.py
View File

@@ -46,6 +46,23 @@ from ipalib.plugins.baseldap import *
from ipalib import _, ngettext
output_params = (
Str('memberuser_user?',
label='Member User',
),
Str('memberuser_group?',
label='Member Group',
),
Str('memberhost_host?',
label=_('Member Host'),
),
Str('memberhost_hostgroup?',
label='Member Hostgroup',
),
Str('externalhost?',
label=_('External host'),
),
)
class netgroup(LDAPObject):
"""
Netgroup object.
@@ -55,13 +72,14 @@ class netgroup(LDAPObject):
object_name_plural = 'netgroups'
object_class = ['ipaobject', 'ipaassociation', 'ipanisnetgroup']
default_attributes = [
'cn', 'description', 'member', 'memberof', 'externalhost',
'nisdomainname',
'cn', 'description', 'memberof', 'externalhost',
'nisdomainname', 'memberuser', 'memberhost',
]
uuid_attribute = 'ipauniqueid'
attribute_members = {
'member': ['user', 'group', 'host', 'hostgroup', 'netgroup'],
'memberof': ['netgroup'],
'memberuser': ['user', 'group'],
'memberhost': ['host', 'hostgroup'],
}
label = _('Net Groups')
@@ -88,26 +106,6 @@ class netgroup(LDAPObject):
doc=_('IPA unique ID'),
flags=['no_create', 'no_update'],
),
Str('member_user?',
label='Member User',
flags=['no_create', 'no_update', 'no_search'],
),
Str('member_group?',
label='Member Group',
flags=['no_create', 'no_update', 'no_search'],
),
Str('member_host?',
label=_('Member host'),
flags=['no_create', 'no_update', 'no_search'],
),
Str('member_hostgroup?',
label='Member Hostgroup',
flags=['no_create', 'no_update', 'no_search'],
),
Str('externalhost?',
label=_('External host'),
flags=['no_create', 'no_update', 'no_search'],
),
)
def get_dn(self, *keys, **kwargs):
@@ -135,6 +133,7 @@ class netgroup_add(LDAPCreate):
"""
Create new netgroup.
"""
has_output_params = output_params
def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
if not dn.startswith('cn='):
msg = 'netgroup with name "%s" already exists' % keys[-1]
@@ -160,6 +159,7 @@ class netgroup_mod(LDAPUpdate):
"""
Modify netgroup.
"""
has_output_params = output_params
api.register(netgroup_mod)
@@ -168,6 +168,7 @@ class netgroup_find(LDAPSearch):
"""
Search the groups.
"""
has_output_params = output_params
api.register(netgroup_find)
@@ -176,6 +177,7 @@ class netgroup_show(LDAPRetrieve):
"""
Display netgroup.
"""
has_output_params = output_params
api.register(netgroup_show)
@@ -184,14 +186,18 @@ class netgroup_add_member(LDAPAddMember):
"""
Add members to netgroup.
"""
has_output_params = output_params
member_attributes = ['memberuser', 'memberhost']
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
if 'member' in failed and 'host' in failed['member']:
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('member', [])
members = entry_attrs.get('memberhost', [])
external_hosts = entry_attrs_.get('externalhost', [])
failed_hosts = []
completed_external = 0
for host in failed['member']['host']:
for host in failed['memberhost']['host']:
host = host.lower()
host_dn = self.api.Object['host'].get_dn(host)
if host not in external_hosts and host_dn not in members:
@@ -204,7 +210,7 @@ class netgroup_add_member(LDAPAddMember):
ldap.update_entry(dn, {'externalhost': external_hosts})
except errors.EmptyModlist:
pass
failed['member']['host'] = failed_hosts
failed['memberhost']['host'] = failed_hosts
entry_attrs['externalhost'] = external_hosts
return (completed + completed_external, dn)
@@ -216,13 +222,17 @@ class netgroup_remove_member(LDAPRemoveMember):
"""
Remove members from netgroup.
"""
has_output_params = output_params
member_attributes = ['memberuser', 'memberhost']
def post_callback(self, ldap, completed, failed, dn, entry_attrs, *keys, **options):
if 'member' in failed and 'host' in failed['member']:
# 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['member']['host']:
for host in failed['memberhost']['host']:
host = host.lower()
if host in external_hosts:
external_hosts.remove(host)
@@ -234,7 +244,7 @@ class netgroup_remove_member(LDAPRemoveMember):
ldap.update_entry(dn, {'externalhost': external_hosts})
except errors.EmptyModlist:
pass
failed['member']['host'] = failed_hosts
failed['memberhost']['host'] = failed_hosts
entry_attrs['externalhost'] = external_hosts
return (completed + completed_external, dn)

146
tests/test_xmlrpc/test_netgroup_plugin.py
View File

@@ -22,10 +22,18 @@ Test the `ipalib/plugins/netgroup.py` module.
"""
import sys
import nose
import krbV
from xmlrpc_test import XMLRPC_test, assert_attr_equal, assert_is_member
from ipalib import api
from ipalib import errors
from ipaserver.plugins.ldap2 import ldap2
# Global so we can save the value between tests
netgroup_dn = None
# See if our LDAP server is up and we can talk to it over GSSAPI
ccache = krbV.default_context().default_ccache().name
class test_netgroup(XMLRPC_test):
"""
@@ -40,7 +48,7 @@ class test_netgroup(XMLRPC_test):
host_localityname = u'Undisclosed location'
host_kw = {'fqdn': host_fqdn, 'description': host_description, 'localityname': host_localityname, 'raw': True}
hg_cn = u'ng1'
hg_cn = u'hg1'
hg_description = u'Netgroup'
hg_kw = {'cn': hg_cn, 'description': hg_description, 'raw': True}
@@ -50,6 +58,13 @@ class test_netgroup(XMLRPC_test):
user_home = u'/home/%s' % user_uid
user_kw = {'givenname': user_givenname,'sn': user_sn,'uid': user_uid,'homedirectory': user_home, 'raw': True}
# user2 is a member of testgroup
user2_uid = u'pexample'
user2_givenname = u'Pete'
user2_sn = u'Example'
user2_home = u'/home/%s' % user2_uid
user2_kw = {'givenname': user2_givenname,'sn': user2_sn,'uid': user2_uid,'homedirectory': user2_home, 'raw': True}
group_cn = u'testgroup'
group_description = u'This is a test'
group_kw = {'description': group_description,'cn': group_cn}
@@ -81,11 +96,22 @@ class test_netgroup(XMLRPC_test):
assert_attr_equal(entry, 'givenname', self.user_givenname)
assert_attr_equal(entry, 'uid', self.user_uid)
# Add our second user
entry = api.Command['user_add'](**self.user2_kw)['result']
assert_attr_equal(entry, 'givenname', self.user2_givenname)
assert_attr_equal(entry, 'uid', self.user2_uid)
# Add a group
entry = api.Command['group_add'](**self.group_kw)['result']
assert_attr_equal(entry, 'description', self.group_description)
assert_attr_equal(entry, 'cn', self.group_cn)
# Add a user to the group
kw = {'raw': True}
kw['user'] = self.user2_uid
res = api.Command['group_add_member'](self.group_cn, **kw)
assert res['completed'] == 1
def test_3_netgroup_add_member(self):
"""
Test the `xmlrpc.netgroup_add_member` method.
@@ -93,25 +119,25 @@ class test_netgroup(XMLRPC_test):
kw = {'raw': True}
kw['host'] = self.host_fqdn
entry = api.Command['netgroup_add_member'](self.ng_cn, **kw)['result']
assert_is_member(entry, 'fqdn=%s' % self.host_fqdn)
assert_is_member(entry, 'fqdn=%s' % self.host_fqdn, 'memberhost')
kw = {'raw': True}
kw['hostgroup'] = self.hg_cn
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 1
assert_is_member(ret['result'], 'cn=%s' % self.hg_cn)
assert_is_member(ret['result'], 'cn=%s' % self.hg_cn, 'memberhost')
kw = {'raw': True}
kw['user'] = self.user_uid
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 1
assert_is_member(ret['result'], 'uid=%s' % self.user_uid)
assert_is_member(ret['result'], 'uid=%s' % self.user_uid, 'memberuser')
kw = {'raw': True}
kw['group'] = self.group_cn
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 1
assert_is_member(ret['result'], 'cn=%s' % self.group_cn)
assert_is_member(ret['result'], 'cn=%s' % self.group_cn, 'memberuser')
def test_4_netgroup_add_member(self):
"""
@@ -122,36 +148,36 @@ class test_netgroup(XMLRPC_test):
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'host' in failed['member']
assert self.host_fqdn in failed['member']['host']
assert 'memberhost' in failed
assert 'host' in failed['memberhost']
assert self.host_fqdn in failed['memberhost']['host']
kw = {'raw': True}
kw['hostgroup'] = self.hg_cn
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'hostgroup' in failed['member']
assert self.hg_cn in failed['member']['hostgroup']
assert 'memberhost' in failed
assert 'hostgroup' in failed['memberhost']
assert self.hg_cn in failed['memberhost']['hostgroup']
kw = {'raw': True}
kw['user'] = self.user_uid
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'user' in failed['member']
assert self.user_uid in failed['member']['user']
assert 'memberuser' in failed
assert 'user' in failed['memberuser']
assert self.user_uid in failed['memberuser']['user']
kw = {'raw': True}
kw['group'] = self.group_cn
ret = api.Command['netgroup_add_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'group' in failed['member']
assert self.group_cn in failed['member']['group']
assert 'memberuser' in failed
assert 'group' in failed['memberuser']
assert self.group_cn in failed['memberuser']['group']
def test_5_netgroup_add_member(self):
"""
@@ -166,28 +192,69 @@ class test_netgroup(XMLRPC_test):
def test_6_netgroup_show(self):
"""
Test the `xmlrpc.netgroup_show` method.
Test the `xmlrpc.netgroup_show` method with --all.
"""
entry = api.Command['netgroup_show'](self.ng_cn, all=True, raw=True)['result']
assert_attr_equal(entry, 'description', self.ng_description)
assert_attr_equal(entry, 'cn', self.ng_cn)
assert_is_member(entry, 'fqdn=%s' % self.host_fqdn)
assert_is_member(entry, 'cn=%s' % self.hg_cn)
assert_is_member(entry, 'uid=%s' % self.user_uid)
assert_is_member(entry, 'cn=%s' % self.group_cn)
assert_is_member(entry, 'fqdn=%s' % self.host_fqdn, 'memberhost')
assert_is_member(entry, 'cn=%s' % self.hg_cn, 'memberhost')
assert_is_member(entry, 'uid=%s' % self.user_uid, 'memberuser')
assert_is_member(entry, 'cn=%s' % self.group_cn, 'memberuser')
assert_attr_equal(entry, 'objectclass', 'ipaobject')
assert_attr_equal(entry, 'objectclass', 'ipanisnetgroup')
assert_attr_equal(entry, 'objectclass', 'ipaassociation')
def test_6a_netgroup_show(self):
"""
Test the `xmlrpc.netgroup_show` method.
"""
global netgroup_dn
entry = api.Command['netgroup_show'](self.ng_cn, all=False, raw=True)['result']
assert_attr_equal(entry, 'description', self.ng_description)
assert_attr_equal(entry, 'cn', self.ng_cn)
assert_is_member(entry, 'fqdn=%s' % self.host_fqdn, 'memberhost')
assert_is_member(entry, 'cn=%s' % self.hg_cn, 'memberhost')
assert_is_member(entry, 'uid=%s' % self.user_uid, 'memberuser')
assert_is_member(entry, 'cn=%s' % self.group_cn, 'memberuser')
netgroup_dn = entry['dn']
def test_6b_netgroup_show(self):
"""
Confirm the underlying triples
"""
# Do an LDAP query to the compat area and verify that the entry
# is correct
conn = ldap2(shared_instance=False, ldap_uri=api.env.ldap_uri, base_dn=api.env.basedn)
conn.connect(ccache=ccache)
try:
entries = conn.find_entries('cn=%s' % self.ng_cn,
base_dn='cn=ng,cn=compat,%s' % api.env.basedn)
except errors.NotFound:
raise nose.SkipTest('compat and nis are not enabled, skipping test')
finally:
conn.disconnect()
triples = entries[0][0][1]['nisnetgrouptriple']
# This may not prove to be reliable since order is not guaranteed
# and even which user gets into which triple can be random.
assert '(nosuchhost,jexample,example.com)' in triples
assert '(ipatesthost.%s,pexample,example.com)' % api.env.domain in triples
def test_7_netgroup_find(self):
"""
Test the `xmlrpc.hostgroup_find` method.
Test the `xmlrpc.netgroup_find` method.
"""
entries = api.Command.netgroup_find(self.ng_cn, raw=True)['result']
result = api.Command.netgroup_find(self.ng_cn, raw=True)
entries = result['result']
assert(result['count'] == 1)
assert_attr_equal(entries[0], 'description', self.ng_description)
assert_attr_equal(entries[0], 'cn', self.ng_cn)
def test_8_netgroup_mod(self):
"""
Test the `xmlrpc.hostgroup_mod` method.
Test the `xmlrpc.netgroup_mod` method.
"""
newdesc = u'Updated host group'
modkw = {'cn': self.ng_cn, 'description': newdesc, 'raw': True}
@@ -201,7 +268,7 @@ class test_netgroup(XMLRPC_test):
def test_9_netgroup_remove_member(self):
"""
Test the `xmlrpc.hostgroup_remove_member` method.
Test the `xmlrpc.netgroup_remove_member` method.
"""
kw = {'raw': True}
kw['host'] = self.host_fqdn
@@ -232,18 +299,18 @@ class test_netgroup(XMLRPC_test):
ret = api.Command['netgroup_remove_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'host' in failed['member']
assert self.host_fqdn in failed['member']['host']
assert 'memberhost' in failed
assert 'host' in failed['memberhost']
assert self.host_fqdn in failed['memberhost']['host']
kw = {'raw': True}
kw['hostgroup'] = self.hg_cn
ret = api.Command['netgroup_remove_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'hostgroup' in failed['member']
assert self.hg_cn in failed['member']['hostgroup']
assert 'memberhost' in failed
assert 'hostgroup' in failed['memberhost']
assert self.hg_cn in failed['memberhost']['hostgroup']
kw = {'raw': True}
kw['user'] = self.user_uid
@@ -251,18 +318,18 @@ class test_netgroup(XMLRPC_test):
ret = api.Command['netgroup_remove_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'user' in failed['member']
assert self.user_uid in failed['member']['user']
assert 'memberuser' in failed
assert 'user' in failed['memberuser']
assert self.user_uid in failed['memberuser']['user']
kw = {'raw': True}
kw['group'] = self.group_cn
ret = api.Command['netgroup_remove_member'](self.ng_cn, **kw)
assert ret['completed'] == 0
failed = ret['failed']
assert 'member' in failed
assert 'group' in failed['member']
assert self.group_cn in failed['member']['group']
assert 'memberuser' in failed
assert 'group' in failed['memberuser']
assert self.group_cn in failed['memberuser']['group']
def test_b_netgroup_del(self):
"""
@@ -304,8 +371,9 @@ class test_netgroup(XMLRPC_test):
else:
assert False
# Remove the user
# Remove the users
assert api.Command['user_del'](self.user_uid)['result'] is True
assert api.Command['user_del'](self.user2_uid)['result'] is True
# Verify that it is gone
try: