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

LGTM: raise handle_not_found()

Browse Source 
Turn calls "handle_not_found()" into "raise handle_not_found()" to
indicate control flow chance. It makes the code easier to understand,
the control flow more obvious and helps static analyzers.

It's OK to raise here because handle_not_found() always raises an
exception.

https://pagure.io/freeipa/issue/7344

Signed-off-by: Christian Heimes <cheimes@redhat.com>
Reviewed-By: Fraser Tweedale <ftweedal@redhat.com>
This commit is contained in:
Christian Heimes 2018-01-03 12:11:15 +01:00
parent 73ee9ff40e
commit f60b2c5906
25 changed files with 205 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
ipaserver/plugins/aci.py
View File

@ -282,7 +282,7 @@ def _make_aci(ldap, current, aciname, kw):
try:
api.Object['group'].get_dn_if_exists(kw['memberof'])
except errors.NotFound:
api.Object['group'].handle_not_found(kw['memberof'])
raise api.Object['group'].handle_not_found(kw['memberof'])
groupdn = _group_from_memberof(kw['memberof'])
a.set_target_filter('memberOf=%s' % groupdn)
if valid['filter']:

2
ipaserver/plugins/automember.py
View File

@ -769,7 +769,7 @@ class automember_rebuild(Method):
try:
obj.get_dn_if_exists(name)
except errors.NotFound:
obj.handle_not_found(name)
raise obj.handle_not_found(name)
search_filter = ldap.make_filter_from_attr(
obj.primary_key.name,
names,

35
ipaserver/plugins/baseldap.py
View File

@ -753,6 +753,10 @@ class LDAPObject(Object):
entry_attrs[attr] = False
def handle_not_found(self, *keys):
"""Handle NotFound exception
Must raise errors.NotFound again.
"""
pkey = ''
if self.primary_key:
pkey = keys[-1]
@ -1015,7 +1019,7 @@ last, after all sets and adds."""),
dn, needldapattrs
)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
# Provide a nice error message when user tries to delete an
# attribute that does not exist on the entry (and user is not
@ -1220,7 +1224,7 @@ class LDAPCreate(BaseLDAPCommand, crud.Create):
entry_attrs = self._exc_wrapper(keys, options, ldap.get_entry)(
entry_attrs.dn, attrs_list)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.obj.get_indirect_members(entry_attrs, attrs_list)
@ -1320,7 +1324,7 @@ class LDAPRetrieve(LDAPQuery):
dn, attrs_list
)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.obj.get_indirect_members(entry_attrs, attrs_list)
@ -1450,7 +1454,7 @@ class LDAPUpdate(LDAPQuery, crud.Update):
# Attempt to rename to the current name, ignore
pass
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
finally:
# Delete the primary_key from entry_attrs either way
del entry_attrs[self.obj.primary_key.name]
@ -1469,7 +1473,7 @@ class LDAPUpdate(LDAPQuery, crud.Update):
if not rdnupdate:
raise e
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
try:
entry_attrs = self._exc_wrapper(keys, options, ldap.get_entry)(
@ -1548,14 +1552,16 @@ class LDAPDelete(LDAPMultiQuery):
for entry_attrs in subentries:
delete_subtree(entry_attrs.dn)
try:
self._exc_wrapper(nkeys, options, ldap.delete_entry)(base_dn)
self._exc_wrapper(nkeys, options, ldap.delete_entry)(
base_dn
)
except errors.NotFound:
self.obj.handle_not_found(*nkeys)
raise self.obj.handle_not_found(*nkeys)
try:
self._exc_wrapper(nkeys, options, ldap.delete_entry)(dn)
except errors.NotFound:
self.obj.handle_not_found(*nkeys)
raise self.obj.handle_not_found(*nkeys)
except errors.NotAllowedOnNonLeaf:
if not self.subtree_delete:
raise
@ -1712,7 +1718,7 @@ class LDAPAddMember(LDAPModMember):
dn, attrs_list
)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.obj.get_indirect_members(entry_attrs, attrs_list)
@ -1813,7 +1819,7 @@ class LDAPRemoveMember(LDAPModMember):
dn, attrs_list
)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.obj.get_indirect_members(entry_attrs, attrs_list)
@ -2055,10 +2061,13 @@ class LDAPSearch(BaseLDAPCommand, crud.Search):
except errors.EmptyResult:
(entries, truncated) = ([], False)
except errors.NotFound:
self.api.Object[self.obj.parent_object].handle_not_found(*keys)
return self.api.Object[self.obj.parent_object].handle_not_found(
*keys)
for callback in self.get_callbacks('post'):
truncated = callback(self, ldap, entries, truncated, *args, **options)
truncated = callback(
self, ldap, entries, truncated, *args, **options
)
if self.sort_result_entries:
if self.obj.primary_key:
@ -2370,7 +2379,7 @@ class BaseLDAPModAttribute(LDAPQuery):
self._exc_wrapper(keys, options, ldap.update_entry)(update)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
try:
entry_attrs = self._exc_wrapper(keys, options, ldap.get_entry)(

2
ipaserver/plugins/baseuser.py
View File

@ -529,7 +529,7 @@ class baseuser_mod(LDAPUpdate):
if 'krbcanonicalname' not in old_entry:
return
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.context.krbprincipalname = old_entry.get(
'krbprincipalname', [])

16
ipaserver/plugins/caacl.py
View File

@ -278,7 +278,7 @@ class caacl_mod(LDAPUpdate):
entry_attrs = ldap.get_entry(dn, attrs_list)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(options, 'ipacacategory') and 'ipamemberca' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_(
@ -332,7 +332,7 @@ class caacl_enable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['TRUE']
@ -361,7 +361,7 @@ class caacl_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['FALSE']
@ -391,7 +391,7 @@ class caacl_add_user(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(entry_attrs, 'usercategory'):
raise errors.MutuallyExclusiveError(
reason=_("users cannot be added when user category='all'"))
@ -423,7 +423,7 @@ class caacl_add_host(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(entry_attrs, 'hostcategory'):
raise errors.MutuallyExclusiveError(
reason=_("hosts cannot be added when host category='all'"))
@ -453,7 +453,7 @@ class caacl_add_service(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(entry_attrs, 'servicecategory'):
raise errors.MutuallyExclusiveError(reason=_(
"services cannot be added when service category='all'"))
@ -493,7 +493,7 @@ class caacl_add_profile(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(entry_attrs, 'ipacertprofilecategory'):
raise errors.MutuallyExclusiveError(reason=_(
"profiles cannot be added when profile category='all'"))
@ -525,7 +525,7 @@ class caacl_add_ca(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(entry_attrs, 'ipacacategory'):
raise errors.MutuallyExclusiveError(reason=_(
"CAs cannot be added when CA category='all'"))

4
ipaserver/plugins/certmap.py
View File

@ -349,7 +349,7 @@ class certmaprule_enable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['TRUE']
@ -378,7 +378,7 @@ class certmaprule_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['FALSE']

2
ipaserver/plugins/config.py
View File

@ -532,7 +532,7 @@ class config_mod(LDAPUpdate):
try:
self.api.Object.server.get_dn_if_exists(new_master)
except errors.NotFound:
self.api.Object.server.handle_not_found(new_master)
raise self.api.Object.server.handle_not_found(new_master)
backend = self.api.Backend.serverroles
backend.config_update(ca_renewal_master_server=new_master)

44
ipaserver/plugins/dns.py
View File

@ -2164,7 +2164,7 @@ class DNSZoneBase_del(LDAPDelete):
def pre_callback(self, ldap, dn, *nkeys, **options):
assert isinstance(dn, DN)
if not _check_DN_objectclass(ldap, dn, self.obj.object_class):
self.obj.handle_not_found(*nkeys)
raise self.obj.handle_not_found(*nkeys)
return dn
def post_callback(self, ldap, dn, *keys, **options):
@ -2227,7 +2227,7 @@ class DNSZoneBase_show(LDAPRetrieve):
def pre_callback(self, ldap, dn, attrs_list, *keys, **options):
assert isinstance(dn, DN)
if not _check_DN_objectclass(ldap, dn, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
return dn
def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
@ -2246,10 +2246,10 @@ class DNSZoneBase_disable(LDAPQuery):
try:
entry = ldap.get_entry(dn, ['idnszoneactive', 'objectclass'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if not _check_entry_objectclass(entry, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
entry['idnszoneactive'] = ['FALSE']
@ -2271,10 +2271,10 @@ class DNSZoneBase_enable(LDAPQuery):
try:
entry = ldap.get_entry(dn, ['idnszoneactive', 'objectclass'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if not _check_entry_objectclass(entry, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
entry['idnszoneactive'] = ['TRUE']
@ -2297,10 +2297,11 @@ class DNSZoneBase_add_permission(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['objectclass'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
else:
if not _check_entry_objectclass(entry_attrs, self.obj.object_class):
self.obj.handle_not_found(*keys)
if not _check_entry_objectclass(
entry_attrs, self.obj.object_class):
raise self.obj.handle_not_found(*keys)
permission_name = self.obj.permission_name(keys[-1])
@ -2353,10 +2354,10 @@ class DNSZoneBase_remove_permission(LDAPQuery):
try:
entry = ldap.get_entry(dn, ['managedby', 'objectclass'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
else:
if not _check_entry_objectclass(entry, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
entry['managedby'] = None
@ -2863,13 +2864,13 @@ class dnszone_mod(DNSZoneBase_mod):
takes_options = DNSZoneBase_mod.takes_options + (
Flag('force',
label=_('Force'),
doc=_('Force nameserver change even if nameserver not in DNS'),
),
doc=_('Force nameserver change even if nameserver not in DNS')),
)
def pre_callback(self, ldap, dn, entry_attrs, attrs_list, *keys, **options):
def pre_callback(self, ldap, dn, entry_attrs, attrs_list,
*keys, **options):
if not _check_DN_objectclass(ldap, dn, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if 'idnssoamname' in entry_attrs:
nameserver = entry_attrs['idnssoamname']
if nameserver:
@ -3146,10 +3147,11 @@ class dnsrecord(LDAPObject):
try:
entry = ldap.get_entry(dn, ['objectclass'])
except errors.NotFound:
parent_object.handle_not_found(zone)
raise parent_object.handle_not_found(zone)
else:
# only master zones can contain records
if 'idnszone' not in [x.lower() for x in entry.get('objectclass', [])]:
if 'idnszone' not in [x.lower()
for x in entry.get('objectclass', [])]:
raise errors.ValidationError(
name='dnszoneidnsname',
error=_(u'only master zones can contain records')
@ -3751,7 +3753,7 @@ class dnsrecord_mod(LDAPUpdate):
try:
old_entry = ldap.get_entry(dn, _record_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if updated_attrs:
for attr in updated_attrs:
@ -3876,7 +3878,7 @@ class dnsrecord_del(LDAPUpdate):
try:
old_entry = ldap.get_entry(dn, _record_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
for attr in entry_attrs.keys():
if attr not in _record_attributes:
@ -4407,10 +4409,10 @@ class dnsforwardzone_mod(DNSZoneBase_mod):
try:
entry = ldap.get_entry(dn)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if not _check_entry_objectclass(entry, self.obj.object_class):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
policy = self.obj.default_forward_policy
forwarders = []

24
ipaserver/plugins/group.py
View File

@ -659,17 +659,27 @@ class group_detach(LDAPQuery):
try:
user_attrs = ldap.get_entry(user_dn)
except errors.NotFound:
self.obj.handle_not_found(*keys)
is_managed = self.obj.has_objectclass(user_attrs['objectclass'], 'mepmanagedentry')
raise self.obj.handle_not_found(*keys)
is_managed = self.obj.has_objectclass(
user_attrs['objectclass'], 'mepmanagedentry'
)
if (not ldap.can_write(user_dn, "objectclass") or
not (ldap.can_write(user_dn, "mepManagedEntry")) and is_managed):
raise errors.ACIError(info=_('not allowed to modify user entries'))
not ldap.can_write(user_dn, "mepManagedEntry")
and is_managed):
raise errors.ACIError(
info=_('not allowed to modify user entries')
)
group_attrs = ldap.get_entry(group_dn)
is_managed = self.obj.has_objectclass(group_attrs['objectclass'], 'mepmanagedby')
is_managed = self.obj.has_objectclass(
group_attrs['objectclass'], 'mepmanagedby'
)
if (not ldap.can_write(group_dn, "objectclass") or
not (ldap.can_write(group_dn, "mepManagedBy")) and is_managed):
raise errors.ACIError(info=_('not allowed to modify group entries'))
not ldap.can_write(group_dn, "mepManagedBy")
and is_managed):
raise errors.ACIError(
info=_('not allowed to modify group entries')
)
objectclasses = user_attrs['objectclass']
try:

46
ipaserver/plugins/hbacrule.py
View File

@ -339,14 +339,24 @@ class hbacrule_mod(LDAPUpdate):
entry_attrs = ldap.get_entry(dn, attrs_list)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(options, 'usercategory') and 'memberuser' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_("user category cannot be set to 'all' while there are allowed users"))
raise errors.MutuallyExclusiveError(
reason=_("user category cannot be set to 'all' while there "
"are allowed users")
)
if is_all(options, 'hostcategory') and 'memberhost' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_("host category cannot be set to 'all' while there are allowed hosts"))
if is_all(options, 'servicecategory') and 'memberservice' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_("service category cannot be set to 'all' while there are allowed services"))
raise errors.MutuallyExclusiveError(
reason=_("host category cannot be set to 'all' while there "
"are allowed hosts")
)
if (is_all(options, 'servicecategory')
and 'memberservice' in entry_attrs):
raise errors.MutuallyExclusiveError(
reason=_("service category cannot be set to 'all' while "
"there are allowed services")
)
return dn
@ -381,7 +391,7 @@ class hbacrule_enable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['TRUE']
@ -411,7 +421,7 @@ class hbacrule_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['FALSE']
@ -453,7 +463,7 @@ class hbacrule_add_accesstime(LDAPQuery):
except errors.EmptyModlist:
pass
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
return dict(result=True)
@ -484,7 +494,7 @@ class hbacrule_remove_accesstime(LDAPQuery):
except (ValueError, errors.EmptyModlist):
pass
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
return dict(result=True)
@ -502,9 +512,9 @@ class hbacrule_add_user(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
if 'usercategory' in entry_attrs and \
entry_attrs['usercategory'][0].lower() == 'all':
raise self.obj.handle_not_found(*keys)
if ('usercategory' in entry_attrs and
entry_attrs['usercategory'][0].lower() == 'all'):
raise errors.MutuallyExclusiveError(
reason=_("users cannot be added when user category='all'"))
return dn
@ -533,9 +543,9 @@ class hbacrule_add_host(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
if 'hostcategory' in entry_attrs and \
entry_attrs['hostcategory'][0].lower() == 'all':
raise self.obj.handle_not_found(*keys)
if ('hostcategory' in entry_attrs and
entry_attrs['hostcategory'][0].lower() == 'all'):
raise errors.MutuallyExclusiveError(
reason=_("hosts cannot be added when host category='all'"))
return dn
@ -588,9 +598,9 @@ class hbacrule_add_service(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
if 'servicecategory' in entry_attrs and \
entry_attrs['servicecategory'][0].lower() == 'all':
raise self.obj.handle_not_found(*keys)
if ('servicecategory' in entry_attrs and
entry_attrs['servicecategory'][0].lower() == 'all'):
raise errors.MutuallyExclusiveError(reason=_(
"services cannot be added when service category='all'"))
return dn

10
ipaserver/plugins/host.py
View File

@ -899,7 +899,7 @@ class host_mod(LDAPUpdate):
try:
entry_attrs_old = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
old_certs = entry_attrs_old.get('usercertificate', [])
removed_certs = set(old_certs) - set(certs)
for cert in removed_certs:
@ -931,7 +931,7 @@ class host_mod(LDAPUpdate):
result = api.Command['dnszone_show'](domain)['result']
domain = result['idnsname'][0]
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
update_sshfp_record(domain, unicode(parts[0]), entry_attrs)
if 'ipasshpubkey' in entry_attrs:
@ -1020,7 +1020,7 @@ class host_find(LDAPSearch):
try:
entry_attrs = ldap.get_entry(dn, ['managedby'])
except errors.NotFound:
self.obj.handle_not_found(pkey)
raise self.obj.handle_not_found(pkey)
hosts.append(set(entry_attrs.get('managedby', '')))
hosts = list(reduce(lambda s1, s2: s1 & s2, hosts))
@ -1037,7 +1037,7 @@ class host_find(LDAPSearch):
try:
entry_attrs = ldap.get_entry(dn, ['managedby'])
except errors.NotFound:
self.obj.handle_not_found(pkey)
raise self.obj.handle_not_found(pkey)
not_hosts += entry_attrs.get('managedby', [])
not_hosts = list(set(not_hosts))
@ -1191,7 +1191,7 @@ class host_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if self.api.Command.ca_is_enabled()['result']:
certs = self.api.Command.cert_find(host=keys)['result']

4
ipaserver/plugins/idrange.py
View File

@ -535,7 +535,7 @@ class idrange_del(LDAPDelete):
'ipaidrangesize',
'ipanttrusteddomainsid'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
# Check whether we leave any object with id in deleted range
old_base_id = int(old_attrs.get('ipabaseid', [0])[0])
@ -645,7 +645,7 @@ class idrange_mod(LDAPUpdate):
try:
old_attrs = ldap.get_entry(dn, ['*'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if old_attrs['iparangetype'][0] == 'ipa-local':
raise errors.ExecutionError(

8
ipaserver/plugins/idviews.py
View File

@ -153,7 +153,7 @@ class idview(LDAPObject):
try:
orig_entry_attrs = ldap.get_entry(dn, ['objectclass'])
except errors.NotFound:
self.handle_not_found(*keys)
raise self.handle_not_found(*keys)
orig_objectclasses = {
o.lower() for o in orig_entry_attrs.get('objectclass', [])}
@ -587,7 +587,7 @@ def resolve_object_to_anchor(ldap, obj_type, obj, fallback_to_ldap):
pass
# No acceptable object was found
api.Object[obj_type].handle_not_found(obj)
raise api.Object[obj_type].handle_not_found(obj)
def resolve_anchor_to_object_name(ldap, obj_type, anchor):
@ -789,12 +789,12 @@ class baseidoverride_del(LDAPDelete):
try:
entry = ldap.get_entry(dn, ['objectclass'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
# If not, treat it as a failed search
for required_oc in self.obj.object_class:
if not self.obj.has_objectclass(entry['objectclass'], required_oc):
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
return dn

12
ipaserver/plugins/netgroup.py
View File

@ -315,11 +315,17 @@ class netgroup_mod(LDAPUpdate):
entry_attrs = ldap.get_entry(dn, attrs_list)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(options, 'usercategory') and 'memberuser' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_("user category cannot be set to 'all' while there are allowed users"))
raise errors.MutuallyExclusiveError(
reason=_("user category cannot be set to 'all' while there "
"are allowed users")
)
if is_all(options, 'hostcategory') and 'memberhost' in entry_attrs:
raise errors.MutuallyExclusiveError(reason=_("host category cannot be set to 'all' while there are allowed hosts"))
raise errors.MutuallyExclusiveError(
reason=_("host category cannot be set to 'all' while there "
"are allowed hosts")
)
return dn

10
ipaserver/plugins/otptoken.py
View File

@ -99,19 +99,24 @@ def _convert_owner(userobj, entry_attrs, options):
entry_attrs['ipatokenowner'] = [userobj.get_primary_key_from_dn(o)
for o in entry_attrs['ipatokenowner']]
def _normalize_owner(userobj, entry_attrs):
owner = entry_attrs.get('ipatokenowner', None)
if owner:
try:
entry_attrs['ipatokenowner'] = userobj._normalize_manager(owner)[0]
entry_attrs['ipatokenowner'] = userobj._normalize_manager(
owner
)[0]
except NotFound:
userobj.handle_not_found(owner)
raise userobj.handle_not_found(owner)
def _check_interval(not_before, not_after):
if not_before and not_after:
return not_before <= not_after
return True
def _set_token_type(entry_attrs, **options):
klasses = [x.lower() for x in entry_attrs.get('objectclass', [])]
for ttype in TOKEN_TYPES:
@ -122,6 +127,7 @@ def _set_token_type(entry_attrs, **options):
if not options.get('all', False) or options.get('pkey_only', False):
entry_attrs.pop('objectclass', None)
@register()
class otptoken(LDAPObject):
"""

4
ipaserver/plugins/permission.py
View File

@ -1061,7 +1061,7 @@ class permission_del(baseldap.LDAPDelete):
try:
entry = ldap.get_entry(dn, attrs_list=self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if not options.get('force'):
self.obj.reject_system(entry)
@ -1105,7 +1105,7 @@ class permission_mod(baseldap.LDAPUpdate):
attrs_list = self.obj.default_attributes
old_entry = ldap.get_entry(dn, attrs_list=attrs_list)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
self.obj.reject_system(old_entry)
self.obj.upgrade_permission(old_entry)

2
ipaserver/plugins/pwpolicy.py
View File

@ -179,7 +179,7 @@ class cosentry_add(LDAPCreate):
try:
result = ldap.get_entry(group_dn, ['objectclass'])
except errors.NotFound:
self.api.Object.group.handle_not_found(keys[-1])
raise self.api.Object.group.handle_not_found(keys[-1])
oc = [x.lower() for x in result['objectclass']]
if 'mepmanagedentry' in oc:

58
ipaserver/plugins/selinuxusermap.py
View File

@ -355,16 +355,24 @@ class selinuxusermap_mod(LDAPUpdate):
try:
_entry_attrs = ldap.get_entry(dn, attrs_list)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
is_to_be_deleted = lambda x: (x in _entry_attrs and x in entry_attrs) and \
entry_attrs[x] == None
def is_to_be_deleted(x):
return (
(x in _entry_attrs and x in entry_attrs)
and entry_attrs[x] is None
)
# makes sure the local members and hbacrule is not set at the same time
# memberuser or memberhost could have been set using --setattr
is_to_be_set = lambda x: ((x in _entry_attrs and _entry_attrs[x] != None) or \
(x in entry_attrs and entry_attrs[x] != None)) and \
not is_to_be_deleted(x)
def is_to_be_set(x):
return (
(
(x in _entry_attrs and _entry_attrs[x] is not None) or
(x in entry_attrs and entry_attrs[x] is not None)
)
and not is_to_be_deleted(x)
)
are_local_members_to_be_set = any(is_to_be_set(attr)
for attr in ('usercategory',
@ -379,18 +387,26 @@ class selinuxusermap_mod(LDAPUpdate):
if are_local_members_to_be_set and is_hbacrule_to_be_set:
raise errors.MutuallyExclusiveError(reason=notboth_err)
if is_all(entry_attrs, 'usercategory') and 'memberuser' in entry_attrs:
raise errors.MutuallyExclusiveError(reason="user category "
"cannot be set to 'all' while there are allowed users")
if is_all(entry_attrs, 'hostcategory') and 'memberhost' in entry_attrs:
raise errors.MutuallyExclusiveError(reason="host category "
"cannot be set to 'all' while there are allowed hosts")
if (is_all(entry_attrs, 'usercategory')
and 'memberuser' in entry_attrs):
raise errors.MutuallyExclusiveError(
reason="user category cannot be set to 'all' while there "
"are allowed users"
)
if (is_all(entry_attrs, 'hostcategory')
and 'memberhost' in entry_attrs):
raise errors.MutuallyExclusiveError(
reason="host category cannot be set to 'all' while there "
"are allowed hosts"
)
if 'ipaselinuxuser' in entry_attrs:
validate_selinuxuser_inlist(ldap, entry_attrs['ipaselinuxuser'])
if 'seealso' in entry_attrs:
entry_attrs['seealso'] = self.obj._normalize_seealso(entry_attrs['seealso'])
entry_attrs['seealso'] = self.obj._normalize_seealso(
entry_attrs['seealso']
)
return dn
def post_callback(self, ldap, dn, entry_attrs, *keys, **options):
@ -457,7 +473,7 @@ class selinuxusermap_enable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['TRUE']
@ -487,7 +503,7 @@ class selinuxusermap_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['FALSE']
@ -516,9 +532,9 @@ class selinuxusermap_add_user(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
if 'usercategory' in entry_attrs and \
entry_attrs['usercategory'][0].lower() == 'all':
raise self.obj.handle_not_found(*keys)
if ('usercategory' in entry_attrs and
entry_attrs['usercategory'][0].lower() == 'all'):
raise errors.MutuallyExclusiveError(
reason=_("users cannot be added when user category='all'"))
if 'seealso' in entry_attrs:
@ -549,9 +565,9 @@ class selinuxusermap_add_host(LDAPAddMember):
entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
dn = entry_attrs.dn
except errors.NotFound:
self.obj.handle_not_found(*keys)
if 'hostcategory' in entry_attrs and \
entry_attrs['hostcategory'][0].lower() == 'all':
raise self.obj.handle_not_found(*keys)
if ('hostcategory' in entry_attrs and
entry_attrs['hostcategory'][0].lower() == 'all'):
raise errors.MutuallyExclusiveError(
reason=_("hosts cannot be added when host category='all'"))
if 'seealso' in entry_attrs:

4
ipaserver/plugins/server.py
View File

@ -227,7 +227,7 @@ class server_mod(LDAPUpdate):
if entry_attrs.get('ipalocation'):
if not ldap.entry_exists(entry_attrs['ipalocation'][0]):
self.api.Object.location.handle_not_found(
raise self.api.Object.location.handle_not_found(
options['ipalocation_location'])
if 'ipalocation' in entry_attrs or 'ipaserviceweight' in entry_attrs:
@ -893,7 +893,7 @@ class server_conncheck(crud.PKQuery):
try:
self.obj.get_dn_if_exists(*keys[:-1])
except errors.NotFound:
self.obj.handle_not_found(keys[-2])
raise self.obj.handle_not_found(keys[-2])
# the user must have the Replication Administrators privilege
privilege = u'Replication Administrators'

2
ipaserver/plugins/serverrole.py
View File

@ -76,7 +76,7 @@ class server_role(Object):
try:
server_obj.get_dn_if_exists(fqdn)
except NotFound:
server_obj.handle_not_found(fqdn)
raise server_obj.handle_not_found(fqdn)
@register()

2
ipaserver/plugins/service.py
View File

@ -698,7 +698,7 @@ class service_mod(LDAPUpdate):
try:
entry_attrs_old = ldap.get_entry(dn, ['usercertificate'])
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
old_certs = entry_attrs_old.get('usercertificate', [])
removed_certs = set(old_certs) - set(certs)
for cert in removed_certs:

2
ipaserver/plugins/stageuser.py
View File

@ -671,7 +671,7 @@ class stageuser_activate(LDAPQuery):
staging_dn, ['*']
)
except errors.NotFound:
self.obj.handle_not_found(*args)
raise self.obj.handle_not_found(*args)
entry_attrs = dict((k.lower(), v) for (k, v) in entry_attrs.items())
# Check it does not exist an active entry with the same RDN

31
ipaserver/plugins/sudorule.py
View File

@ -417,7 +417,7 @@ class sudorule_mod(LDAPUpdate):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
error = _("%(type)s category cannot be set to 'all' "
"while there are allowed %(objects)s")
@ -487,7 +487,7 @@ class sudorule_enable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['TRUE']
@ -510,7 +510,7 @@ class sudorule_disable(LDAPQuery):
try:
entry_attrs = ldap.get_entry(dn, ['ipaenabledflag'])
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
entry_attrs['ipaenabledflag'] = ['FALSE']
@ -535,7 +535,7 @@ class sudorule_add_allow_command(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(_entry_attrs, 'cmdcategory'):
raise errors.MutuallyExclusiveError(
@ -586,7 +586,7 @@ class sudorule_add_user(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(_entry_attrs, 'usercategory'):
raise errors.MutuallyExclusiveError(
@ -640,7 +640,7 @@ class sudorule_add_host(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if is_all(_entry_attrs, 'hostcategory'):
raise errors.MutuallyExclusiveError(
@ -654,10 +654,11 @@ class sudorule_add_host(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if 'hostmask' in options:
norm = lambda x: unicode(netaddr.IPNetwork(x).cidr)
def norm(x):
return unicode(netaddr.IPNetwork(x).cidr)
old_masks = set(norm(m) for m in _entry_attrs.get('hostmask', []))
new_masks = set(norm(m) for m in options['hostmask'])
@ -699,7 +700,7 @@ class sudorule_remove_host(LDAPRemoveMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if 'hostmask' in options:
def norm(x):
@ -745,7 +746,7 @@ class sudorule_add_runasuser(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if any((is_all(_entry_attrs, 'ipasudorunasusercategory'),
is_all(_entry_attrs, 'ipasudorunasgroupcategory'))):
@ -860,9 +861,9 @@ class sudorule_add_runasgroup(LDAPAddMember):
try:
_entry_attrs = ldap.get_entry(dn, self.obj.default_attributes)
except errors.NotFound:
self.obj.handle_not_found(*keys)
if is_all(_entry_attrs, 'ipasudorunasusercategory') or \
is_all(_entry_attrs, 'ipasudorunasgroupcategory'):
raise self.obj.handle_not_found(*keys)
if (is_all(_entry_attrs, 'ipasudorunasusercategory') or
is_all(_entry_attrs, 'ipasudorunasgroupcategory')):
raise errors.MutuallyExclusiveError(
reason=_("users cannot be added when runAs user or runAs "
"group category='all'"))
@ -943,7 +944,7 @@ class sudorule_add_option(LDAPQuery):
except errors.EmptyModlist:
pass
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
attrs_list = self.obj.default_attributes
entry_attrs = ldap.get_entry(dn, attrs_list)
@ -993,7 +994,7 @@ class sudorule_remove_option(LDAPQuery):
value=options['ipasudoopt']
)
except errors.NotFound:
self.obj.handle_not_found(cn)
raise self.obj.handle_not_found(cn)
attrs_list = self.obj.default_attributes
entry_attrs = ldap.get_entry(dn, attrs_list)

16
ipaserver/plugins/trust.py
View File

@ -590,7 +590,7 @@ class trust(LDAPObject):
ldap.SCOPE_SUBTREE, trustfilter, ['']
)
except errors.NotFound:
self.handle_not_found(keys[-1])
raise self.handle_not_found(keys[-1])
if len(result) > 1:
raise errors.OnlyOneValueAllowed(attr='trust domain')
@ -1273,7 +1273,7 @@ class trustconfig(LDAPObject):
try:
self.backend.get_entry(dn)
except errors.NotFound:
self.api.Object['group'].handle_not_found(group)
raise self.api.Object['group'].handle_not_found(group)
# DN is valid, we can just return
return
except ValueError:
@ -1288,7 +1288,7 @@ class trustconfig(LDAPObject):
[''],
DN(self.api.env.container_group, self.api.env.basedn))
except errors.NotFound:
self.api.Object['group'].handle_not_found(group)
raise self.api.Object['group'].handle_not_found(group)
else:
entry_attrs['ipantfallbackprimarygroup'] = [group_entry.dn]
@ -1645,7 +1645,7 @@ class trustdomain_del(LDAPDelete):
name='domain',
error=_("cannot delete root domain of the trust, "
"use trust-del to delete the trust itself"))
self.obj.handle_not_found(keys[0], domain)
raise self.obj.handle_not_found(keys[0], domain)
try:
self.api.Command.trustdomain_enable(keys[0], domain)
@ -1808,7 +1808,7 @@ class trustdomain_enable(LDAPQuery):
trust_dn = self.obj.get_dn(keys[0], trust_type=u'ad')
trust_entry = ldap.get_entry(trust_dn)
except errors.NotFound:
self.api.Object[self.obj.parent_object].handle_not_found(
raise self.api.Object[self.obj.parent_object].handle_not_found(
keys[0])
dn = self.obj.get_dn(keys[0], keys[1], trust_type=u'ad')
@ -1821,7 +1821,7 @@ class trustdomain_enable(LDAPQuery):
else:
raise errors.AlreadyActive()
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
return dict(
result=True,
@ -1850,7 +1850,7 @@ class trustdomain_disable(LDAPQuery):
trust_dn = self.obj.get_dn(keys[0], trust_type=u'ad')
trust_entry = ldap.get_entry(trust_dn)
except errors.NotFound:
self.api.Object[self.obj.parent_object].handle_not_found(
raise self.api.Object[self.obj.parent_object].handle_not_found(
keys[0])
dn = self.obj.get_dn(keys[0], keys[1], trust_type=u'ad')
@ -1863,7 +1863,7 @@ class trustdomain_disable(LDAPQuery):
else:
raise errors.AlreadyInactive()
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
return dict(
result=True,

8
ipaserver/plugins/user.py
View File

@ -654,7 +654,7 @@ class user_del(baseuser_del):
original_entry_attrs = self._exc_wrapper(
pkey, options, ldap.get_entry)(dn, ['dn'])
except errors.NotFound:
self.obj.handle_not_found(pkey)
raise self.obj.handle_not_found(pkey)
for callback in self.get_callbacks('pre'):
dn = callback(self, ldap, dn, pkey, **options)
@ -710,7 +710,7 @@ class user_del(baseuser_del):
try:
remove_ipaobject_overrides(self.obj.backend, self.obj.api, dn)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if dn.endswith(DN(self.obj.delete_container_dn, api.env.basedn)):
return dn
@ -878,7 +878,7 @@ class user_undel(LDAPQuery):
try:
self._exc_wrapper(keys, options, ldap.get_entry)(delete_dn)
except errors.NotFound:
self.obj.handle_not_found(*keys)
raise self.obj.handle_not_found(*keys)
if delete_dn.endswith(DN(self.obj.active_container_dn,
api.env.basedn)):
raise errors.InvocationError(
@ -1160,7 +1160,7 @@ class user_status(LDAPQuery):
entries.append(newresult)
count += 1
except errors.NotFound:
self.api.Object.user.handle_not_found(*keys)
raise self.api.Object.user.handle_not_found(*keys)
except Exception as e:
logger.error("user_status: Retrieving status for %s failed "
"with %s", dn, str(e))